hunt.io.channel.posix.AbstractStream source code

1 module hunt.io.channel.posix.AbstractStream;
2 
3 // dfmt off
4 version(Posix):
5 // dfmt on
6 
7 import hunt.collection.BufferUtils;
8 import hunt.collection.ByteBuffer;
9 import hunt.event.selector.Selector;
10 import hunt.Functions;
11 import hunt.io.channel.AbstractSocketChannel;
12 import hunt.io.channel.Common;
13 import hunt.io.IoError;
14 import hunt.logging.ConsoleLogger;
15 import hunt.system.Error;
16 
17 import std.format;
18 import std.socket;
19 
20 import core.atomic;
21 import core.stdc.errno;
22 import core.stdc.string;
23 import core.sys.posix.sys.socket : accept;
24 import core.sys.posix.unistd;
25 
26 enum string ResponseData = "HTTP/1.1 200 OK\r\nContent-Length: 13\r\nConnection: Keep-Alive\r\nContent-Type: text/plain\r\nServer: Hunt/1.0\r\nDate: Wed, 17 Apr 2013 12:00:00 GMT\r\n\r\nHello, World!";
27 
28 /**
29 TCP Peer
30 */
31 abstract class AbstractStream : AbstractSocketChannel {
32     enum BufferSize = 4096;
33     private const(ubyte)[] _readBuffer;
34     private ByteBuffer writeBuffer;
35 
36     /**
37     * Warning: The received data is stored a inner buffer. For a data safe,
38     * you would make a copy of it.
39     */
40     protected DataReceivedHandler dataReceivedHandler;
41     protected SimpleEventHandler disconnectionHandler;
42     protected SimpleActionHandler dataWriteDoneHandler;
43 
44     protected AddressFamily _family;
45     protected ByteBuffer _bufferForRead;
46     protected WritingBufferQueue _writeQueue;
47     protected bool isWriteCancelling = false;
48 
49     this(Selector loop, AddressFamily family = AddressFamily.INET, size_t bufferSize = 4096 * 2) {
50         this._family = family;
51         _bufferForRead = BufferUtils.allocate(bufferSize);
52         _bufferForRead.limit(cast(int)bufferSize);
53         _readBuffer = cast(ubyte[])_bufferForRead.array();
54         _writeQueue = new WritingBufferQueue();
55         super(loop, ChannelType.TCP);
56         setFlag(ChannelFlag.Read, true);
57         setFlag(ChannelFlag.Write, true);
58         setFlag(ChannelFlag.ETMode, true);
59     }
60 
61     abstract bool isConnected() nothrow;
62     abstract protected void onDisconnected();
63 
64     /**
65      *
66      */
67     protected bool tryRead() {
68         bool isDone = true;
69         this.clearError();
70         ptrdiff_t len = read(this.handle, cast(void*) _readBuffer.ptr, _readBuffer.length);
71 
72         // ubyte[] rb = new ubyte[BufferSize];
73         // ptrdiff_t len = read(this.handle, cast(void*) rb.ptr, rb.length);
74         version (HUNT_IO_DEBUG) {
75             tracef("reading[fd=%d]: %d bytes", this.handle, len);
76             if (len <= 32)
77                 infof("fd: %d, %d bytes: %(%02X %)", this.handle, len, _readBuffer[0 .. len]);
78             else
79                 infof("fd: %d, 32/%d bytes: %(%02X %)", this.handle, len, _readBuffer[0 .. 32]);
80         }
81 
82         if (len > 0) {
83             if (dataReceivedHandler !is null) {
84                 _bufferForRead.limit(cast(int)len);
85                 _bufferForRead.position(0);
86                 dataReceivedHandler(_bufferForRead);
87 
88                 // ByteBuffer bb = BufferUtils.wrap(cast(byte[])rb[0..len]);
89                 // dataReceivedHandler(bb);
90             }
91 
92             // size_t nBytes = tryWrite(cast(ubyte[])ResponseData);
93 
94             // if(nBytes < ResponseData.length) {
95             //     warning("data lost");
96             // }
97 
98             // It's prossible that there are more data waitting for read in the read I/O space.
99             if (len == _readBuffer.length) {
100                 version (HUNT_IO_DEBUG) infof("Read buffer is full read %d bytes. Need to read again.", len);
101                 isDone = false;
102             }
103         } else if (len == Socket.ERROR) {
104             // https://stackoverflow.com/questions/14595269/errno-35-eagain-returned-on-recv-call
105             // FIXME: Needing refactor or cleanup -@Administrator at 2018-5-8 16:06:13
106             // check more error status
107             this._error = errno != EINTR && errno != EAGAIN && errno != EWOULDBLOCK;
108             if (_error) {
109                 this._errorMessage = getErrorMessage(errno);
110                 errorOccurred(ErrorCode.INTERRUPTED , "read buffer error");
111             } else {
112                 debug warningf("warning on read: fd=%d, errno=%d, message=%s", this.handle,
113                         errno, getErrorMessage(errno));
114             }
115 
116             if(errno == ECONNRESET) {
117                 // https://stackoverflow.com/questions/1434451/what-does-connection-reset-by-peer-mean
118                 onDisconnected();
119                 errorOccurred(ErrorCode.CONNECTIONEESET , "connection reset by peer");
120             }
121         }
122         else {
123             version (HUNT_DEBUG)
124                 infof("connection broken: %s, fd:%d", _remoteAddress.toString(), this.handle);
125             onDisconnected();
126         }
127 
128         return isDone;
129     }
130 
131     override protected void doClose() {
132         version (HUNT_IO_DEBUG) {
133             infof("peer socket closing: fd=%d", this.handle);
134         }
135         if(this.socket is null) {
136             import core.sys.posix.unistd;
137             core.sys.posix.unistd.close(this.handle);
138         } else {
139             this.socket.shutdown(SocketShutdown.BOTH);
140             this.socket.close();
141         }
142 
143         version (HUNT_IO_DEBUG) {
144             infof("peer socket closed: fd=%d", this.handle);
145         }
146     }
147 
148 
149     /**
150      * Try to write a block of data.
151      */
152     protected ptrdiff_t tryWrite(const ubyte[] data) {
153         clearError();
154         // const nBytes = this.socket.send(data);
155         version (HUNT_IO_DEBUG)
156             tracef("try to write: %d bytes, fd=%d", data.length, this.handle);
157         const nBytes = write(this.handle, data.ptr, data.length);
158         // version (HUNT_IO_DEBUG)
159             // tracef("actually written: %d / %d bytes, fd=%d", nBytes, data.length, this.handle);
160 
161         if (nBytes > 0) {
162             return nBytes;
163         }
164 
165         if (nBytes == Socket.ERROR) {
166             // FIXME: Needing refactor or cleanup -@Administrator at 2018-5-8 16:07:38
167             // check more error status
168             // EPIPE/Broken pipe:
169             // https://stackoverflow.com/questions/6824265/sigpipe-broken-pipe
170             // https://github.com/angrave/SystemProgramming/wiki/Networking%2C-Part-7%3A-Nonblocking-I-O%2C-select%28%29%2C-and-epoll
171 
172             if(errno == EAGAIN) {
173                 version (HUNT_IO_DEBUG) {
174                     warningf("warning on write: fd=%d, errno=%d, message=%s", this.handle,
175                         errno, getErrorMessage(errno));
176                 }
177             } else if(errno == EINTR || errno == EWOULDBLOCK) {
178                 // https://stackoverflow.com/questions/38964745/can-a-socket-become-writeable-after-an-ewouldblock-but-before-an-epoll-wait
179                 debug warningf("warning on write: fd=%d, errno=%d, message=%s", this.handle,
180                         errno, getErrorMessage(errno));
181                 // eventLoop.update(this);
182             } else {
183                 this._error = true;
184                 this._errorMessage = getErrorMessage(errno);
185                 if(errno == ECONNRESET) {
186                     // https://stackoverflow.com/questions/1434451/what-does-connection-reset-by-peer-mean
187                     onDisconnected();
188                     errorOccurred(ErrorCode.CONNECTIONEESET , "connection reset by peer");
189                 }
190             }
191         } else {
192             version (HUNT_DEBUG) {
193                 warningf("nBytes=%d, message: %s", nBytes, lastSocketError());
194                 assert(false, "Undefined behavior!");
195             } else {
196                 this._error = true;
197             }
198         }
199 
200         //if (this._error) {
201         //    this._errorMessage = getErrorMessage(errno);
202         //    string msg = format("Socket error on write: fd=%d, code: %d, message=%s",
203         //            this.handle, errno, this.errorMessage);
204         //    debug errorf(msg);
205         //    errorOccurred(msg);
206         //}
207 
208         return 0;
209     }
210 
211     private bool tryNextWrite(ByteBuffer buffer) {
212         const(ubyte)[] data = cast(const(ubyte)[])buffer.getRemaining();
213         version (HUNT_IO_DEBUG) {
214             tracef("writting from a buffer [fd=%d], %d bytes, buffer: %s",
215                 this.handle, data.length, buffer.toString());
216         }
217 
218         ptrdiff_t remaining = data.length;
219         if(data.length == 0)
220             return true;
221 
222         while(remaining > 0 && !_error && !isClosing() && !isWriteCancelling) {
223             ptrdiff_t nBytes = tryWrite(data);
224             version (HUNT_IO_DEBUG)
225             {
226                 tracef("write out once: fd=%d, %d / %d bytes, remaining: %d buffer: %s",
227                     this.handle, nBytes, data.length, remaining, buffer.toString());
228             }
229 
230             if (nBytes > 0) {
231                 remaining -= nBytes;
232                 data = data[nBytes .. $];
233             }
234         }
235 
236         version (HUNT_IO_DEBUG) {
237             if(remaining == 0) {
238                     tracef("A buffer is written out. fd=%d", this.handle);
239                 return true;
240             } else {
241                 warningf("Writing cancelled or an error ocurred. fd=%d", this.handle);
242                 return false;
243             }
244         } else {
245             return remaining == 0;
246         }
247     }
248 
249     void resetWriteStatus() {
250         _writeQueue.clear();
251         atomicStore(_isWritting, false);
252         isWriteCancelling = false;
253     }
254 
255     /**
256      * Should be thread-safe.
257      */
258     override void onWrite() {
259         version (HUNT_IO_DEBUG)
260         {
261             tracef("checking status, isWritting: %s, writeBuffer: %s",
262                 _isWritting, writeBuffer is null ? "null" : writeBuffer.toString());
263         }
264 
265         if(!_isWritting) {
266             version (HUNT_IO_DEBUG)
267             infof("No data needs to be written out. fd=%d", this.handle);
268             return;
269         }
270 
271         if(isClosing() && isWriteCancelling) {
272             version (HUNT_DEBUG) infof("Write cancelled or closed, fd=%d", this.handle);
273             resetWriteStatus();
274             return;
275         }
276 
277         // keep thread-safe here
278         if(!cas(&_isBusyWritting, false, true)) {
279             // version (HUNT_IO_DEBUG)
280             warningf("busy writing. fd=%d", this.handle);
281             return;
282         }
283 
284         scope(exit) {
285             _isBusyWritting = false;
286         }
287 
288         if(writeBuffer !is null) {
289             if(tryNextWrite(writeBuffer)) {
290                 writeBuffer = null;
291             } else {
292                 version (HUNT_IO_DEBUG)
293                 {
294                     infof("waiting to try again... fd=%d, writeBuffer: %s",
295                         this.handle, writeBuffer.toString());
296                 }
297                 // eventLoop.update(this);
298                 return;
299             }
300             version (HUNT_IO_DEBUG)
301                 tracef("running here, fd=%d", this.handle);
302         }
303 
304         if(checkAllWriteDone()) {
305             return;
306         }
307 
308         version (HUNT_IO_DEBUG)
309         {
310             tracef("start to write [fd=%d], writeBuffer %s empty", this.handle, writeBuffer is null ? "is" : "is not");
311         }
312 
313         if(_writeQueue.tryDequeue(writeBuffer)) {
314             if(tryNextWrite(writeBuffer)) {
315                 writeBuffer = null;
316                 checkAllWriteDone();
317             } else {
318             version (HUNT_IO_DEBUG)
319                 infof("waiting to try again: fd=%d, writeBuffer: %s", this.handle, writeBuffer.toString());
320 
321                 // eventLoop.update(this);
322             }
323             version (HUNT_IO_DEBUG)
324                 warningf("running here, fd=%d", this.handle);
325         }
326     }
327     private shared bool _isBusyWritting = false;
328 
329     protected bool checkAllWriteDone() {
330         version (HUNT_IO_DEBUG) {
331             import std.conv;
332             tracef("checking remaining: fd=%d, writeQueue empty: %s", this.handle,
333                 _writeQueue.isEmpty().to!string());
334         }
335 
336         if(_writeQueue.isEmpty()) {
337             resetWriteStatus();
338             version (HUNT_IO_DEBUG)
339                 infof("All data are written out: fd=%d", this.handle);
340             if(dataWriteDoneHandler !is null)
341                 dataWriteDoneHandler(this);
342             return true;
343         }
344 
345         return false;
346     }
347 
348     protected bool doConnect(Address addr) {
349         try {
350             this.socket.connect(addr);
351         } catch (SocketOSException e)
352         {
353             return false;
354         }
355         return true;
356     }
357 
358     void cancelWrite() {
359         isWriteCancelling = true;
360     }
361 }