feat: modify extension internals

- instate operation error checks in php_mrloop_writev_cb and php_mrloop_readv_cb callbacks
- register global TCP connection data buffer size
- add connection data size and connection count parameters to tcpServer() function
- duplicate client connection file descriptor in php_mrloop_tcp_server_recv callback
- add vector count and offset parameters to addReadStream() function
- add vector count parameter to addWriteStream() function
- change DEFAULT_STREAM_BUFF_LEN and DEFAULT_CONN_BUFF_LEN values
- define DEFAULT_VECTOR_COUNT, DEFAULT_READV_OFFSET and PHP_MRLOOP_MAX_TCP_CONNECTIONS

Author: ace411

README.md

@@ -47,14 +47,21 @@ class Mrloop
   public addReadStream(
     resource $stream,
     ?int $nbytes,
+    ?int $vcount,
+    ?int $offset,
     callable $callback,
   ): void
   public addWriteStream(
     resource $stream,
     string $contents,
     callable $callback,
   ): void
-  public tcpServer(int $port, callable $callback): void
+  public tcpServer(
+    int $port,
+    ?int $connections,
+    ?int $nbytes,
+    callable $callback,
+  ): void
   public writev(int|resource $fd, string $message): void
   public static parseHttpRequest(string $request, int $headerlimit = 100): iterable
   public static parseHttpResponse(string $response, int $headerlimit = 100): iterable
@@ -133,6 +140,8 @@ Hello, user
 public Mrloop::addReadStream(
   resource $stream,
   ?int $nbytes,
+  ?int $vcount,
+  ?int $offset,
   callable $callback,
 ): void
 ```
@@ -144,7 +153,12 @@ Funnels file descriptor in readable stream into event loop and thence executes a
 - **stream** (resource) - A userspace-defined readable stream.
   > The file descriptor in the stream is internally given a non-blocking disposition.
 - **nbytes** (int|null) - The number of bytes to read.
-  > Specifying `null` will condition the use of an 8KB buffer.
+  > Specifying `null` will condition the use of a 1KB buffer.
+- **vcount** (int|null) - The number of read vectors to use.
+  > Specifying `null` will condition the use of 2 vectors.
+  > Any value north of `8` will likely result in an inefficient read.
+- **offset** (int|null) - The point at which to start the read operation.
+  > Specifying `null` will condition the use of an offset of `0`.
 - **callback** (callable) - The binary function through which the file's contents and read result code are propagated.
 
 **Return value(s)**
@@ -158,13 +172,15 @@ $loop = Mrloop::init();
 
 $loop->addReadStream(
   $fd = \fopen('/path/to/file', 'r'),
-  null,
+  null, // 1024
+  null, // 2
+  null, // 0
   function (string $contents, int $res) use ($fd) {
     if ($res === 0) {
       echo \sprintf("%s\n", $contents);
-
-      \fclose($fd);
     }
+
+    \fclose($fd);
   },
 );
 
@@ -184,6 +200,7 @@ File contents...
 public Mrloop::addWriteStream(
   resource $stream,
   string $contents,
+  ?int $vcount,
   callable $callback,
 ): void
 ```
@@ -195,6 +212,9 @@ Funnels file descriptor in writable stream into event loop and thence executes a
 - **stream** (resource) - A userspace-defined writable stream.
   > The file descriptor in the stream is internally given a non-blocking disposition.
 - **contents** (string) - The contents to write to the file descriptor.
+- **vcount** (int|null) - The number of write vectors to use.
+  > Specifying `null` will condition the use of 2 vectors.
+  > Any value north of `8` will likely result in an inefficient write.
 - **callback** (callable) - The unary function through which the number of written bytes is propagated.
 
 **Return value(s)**
@@ -211,6 +231,7 @@ $file = '/path/to/file';
 $loop->addWriteStream(
   $fd = \fopen($file, 'w'),
   "file contents...\n",
+  null,
   function (int $nbytes) use ($fd, $file) {
     echo \sprintf("Wrote %d bytes to %s\n", $nbytes, $file);
 
@@ -231,14 +252,25 @@ Wrote 18 bytes to /path/to/file
 ### `Mrloop::tcpServer`
 
 ```php
-public Mrloop::tcpServer(int $port, callable $callback): void
+public Mrloop::tcpServer(
+  int $port,
+  ?int $connections,
+  ?int $nbytes,
+  callable $callback,
+): void
 ```
 
 Instantiates a simple TCP server.
 
 **Parameter(s)**
 
 - **port** (int) - The port on which to listen for incoming connections.
+- **connections** (int|null) - The maximum number of connections to accept.
+  > This parameter does not have any effect when a version of mrloop in which the `mr_tcp_server` function lacks the `max_conn` parameter is included in the compilation process.
+  > Specifying `null` will condition the use of a `1024` connection threshold.
+- **nbytes** (int|null) - The maximum number of readable bytes for each connection.
+  > This setting is akin to the `client_max_body_size` option in NGINX.
+  > Specifying null will condition the use of an `8192` byte threshold.
 - **callback** (callable) - The binary function with which to define a response to a client.
   > Refer to the segment to follow for more information on the callback.
   - **Callback parameters**
@@ -259,6 +291,8 @@ $loop = Mrloop::init();
 
 $loop->tcpServer(
   8080,
+  null,
+  null,
   function (string $message, iterable $client) {
     // print access log
     echo \sprintf(
@@ -311,6 +345,7 @@ $loop = Mrloop::init();
 
 $loop->tcpServer(
   8080,
+  null,
   function (string $message, iterable $client) use ($loop) {
     [
       'client_addr' => $addr,

mrloop_arginfo.h

@@ -18,8 +18,10 @@ ZEND_ARG_TYPE_INFO(0, interval, IS_DOUBLE, 0)
 ZEND_ARG_TYPE_INFO(0, callback, IS_CALLABLE, 0)
 ZEND_END_ARG_INFO()
 
-ZEND_BEGIN_ARG_INFO_EX(arginfo_class_Mrloop_tcpServer, 0, 0, 2)
+ZEND_BEGIN_ARG_INFO_EX(arginfo_class_Mrloop_tcpServer, 0, 0, 4)
 ZEND_ARG_TYPE_INFO(0, port, IS_LONG, 0)
+ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, connections, IS_LONG, 0, "null")
+ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, nbytes, IS_LONG, 0, "null")
 ZEND_ARG_TYPE_INFO(0, callback, IS_CALLABLE, 0)
 ZEND_END_ARG_INFO()
 
@@ -38,15 +40,18 @@ ZEND_ARG_TYPE_INFO(0, signal, IS_LONG, 0)
 ZEND_ARG_TYPE_INFO(0, callback, IS_CALLABLE, 0)
 ZEND_END_ARG_INFO()
 
-ZEND_BEGIN_ARG_INFO_EX(arginfo_class_Mrloop_addReadStream, 0, 0, 3)
+ZEND_BEGIN_ARG_INFO_EX(arginfo_class_Mrloop_addReadStream, 0, 0, 5)
 ZEND_ARG_TYPE_INFO(0, stream, IS_RESOURCE, 0)
 ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, nbytes, IS_LONG, 0, "null")
+ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, vcount, IS_LONG, 0, "null")
+ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, offset, IS_LONG, 0, "null")
 ZEND_ARG_TYPE_INFO(0, callback, IS_CALLABLE, 0)
 ZEND_END_ARG_INFO()
 
-ZEND_BEGIN_ARG_INFO_EX(arginfo_class_Mrloop_addWriteStream, 0, 0, 3)
+ZEND_BEGIN_ARG_INFO_EX(arginfo_class_Mrloop_addWriteStream, 0, 0, 4)
 ZEND_ARG_TYPE_INFO(0, stream, IS_RESOURCE, 0)
 ZEND_ARG_TYPE_INFO(0, contents, IS_STRING, 0)
+ZEND_ARG_TYPE_INFO_WITH_DEFAULT_VALUE(0, vcount, IS_LONG, 0, "null")
 ZEND_ARG_TYPE_INFO(0, callback, IS_CALLABLE, 0)
 ZEND_END_ARG_INFO()
 

src/loop.c

@@ -192,6 +192,11 @@ static void php_mrloop_add_future_tick(INTERNAL_FUNCTION_PARAMETERS)
 
 static void php_mrloop_readv_cb(void *data, int res)
 {
+  if (res < 0)
+  {
+    PHP_MRLOOP_THROW(strerror(-res));
+  }
+
   php_mrloop_cb_t *cb;
   php_iovec_t *iov;
   zval args[2], result;
@@ -222,6 +227,11 @@ static void php_mrloop_readv_cb(void *data, int res)
 }
 static void php_mrloop_writev_cb(void *data, int res)
 {
+  if (res < 0)
+  {
+    PHP_MRLOOP_THROW(strerror(-res));
+  }
+
   php_mrloop_cb_t *cb = (php_mrloop_cb_t *)data;
   zval args[1], result;
   ZVAL_LONG(&args[0], res);
@@ -251,8 +261,9 @@ static void *php_mrloop_tcp_client_setup(int fd, char **buffer, int *bsize)
 
   conn = emalloc(sizeof(php_mrloop_conn_t));
   conn->fd = fd;
+  conn->buffer = emalloc(MRLOOP_G(tcp_buff_size));
   *buffer = conn->buffer;
-  *bsize = DEFAULT_CONN_BUFF_LEN;
+  *bsize = MRLOOP_G(tcp_buff_size);
 
   socklen = sizeof(php_sockaddr_t);
 
@@ -277,6 +288,7 @@ static int php_mrloop_tcp_server_recv(void *conn, int fd, ssize_t nbytes, char *
   {
     mr_close(loop, client->fd);
     efree(client->addr);
+    efree(client->buffer);
     efree(client);
 
     return 1;
@@ -288,7 +300,7 @@ static int php_mrloop_tcp_server_recv(void *conn, int fd, ssize_t nbytes, char *
   array_init(&args[1]);
   add_assoc_string(&args[1], "client_addr", (char *)client->addr);
   add_assoc_long(&args[1], "client_port", client->port);
-  add_assoc_long(&args[1], "client_fd", client->fd);
+  add_assoc_long(&args[1], "client_fd", dup(client->fd));
 
   MRLOOP_G(tcp_cb)->fci.retval = &result;
   MRLOOP_G(tcp_cb)->fci.param_count = 2;
@@ -321,24 +333,39 @@ static void php_mrloop_tcp_server_listen(INTERNAL_FUNCTION_PARAMETERS)
   php_mrloop_t *this;
   zend_fcall_info fci;
   zend_fcall_info_cache fci_cache;
-  zend_long port;
+  zend_long port, max_conn, nbytes;
+  bool max_conn_null, nbytes_null;
+  size_t nconn, fnbytes;
 
   obj = getThis();
   fci = empty_fcall_info;
   fci_cache = empty_fcall_info_cache;
+  max_conn_null = true;
+  nbytes_null = true;
 
-  ZEND_PARSE_PARAMETERS_START(2, 2)
+  ZEND_PARSE_PARAMETERS_START(4, 4)
   Z_PARAM_LONG(port)
+  Z_PARAM_LONG_OR_NULL(max_conn, max_conn_null)
+  Z_PARAM_LONG_OR_NULL(nbytes, nbytes_null)
   Z_PARAM_FUNC(fci, fci_cache)
   ZEND_PARSE_PARAMETERS_END();
 
   this = PHP_MRLOOP_OBJ(obj);
 
+  fnbytes = (size_t)(nbytes_null == true ? DEFAULT_CONN_BUFF_LEN : nbytes);
+  MRLOOP_G(tcp_buff_size) = fnbytes;
+
   MRLOOP_G(tcp_cb) = emalloc(sizeof(php_mrloop_cb_t));
   PHP_CB_TO_MRLOOP_CB(MRLOOP_G(tcp_cb), fci, fci_cache);
   MRLOOP_G(tcp_cb)->data = this->loop;
 
+  nconn = (size_t)(max_conn_null == true ? PHP_MRLOOP_MAX_TCP_CONNECTIONS : (max_conn == 0 ? PHP_MRLOOP_MAX_TCP_CONNECTIONS : max_conn));
+
+#ifdef MRLOOP_H
+  mr_tcp_server(this->loop, (int)port, nconn, php_mrloop_tcp_client_setup, php_mrloop_tcp_server_recv);
+#else
   mr_tcp_server(this->loop, (int)port, php_mrloop_tcp_client_setup, php_mrloop_tcp_server_recv);
+#endif
 
   return;
 }
@@ -528,21 +555,25 @@ static void php_mrloop_add_read_stream(INTERNAL_FUNCTION_PARAMETERS)
   php_iovec_t *iov;
   zend_fcall_info fci;
   zend_fcall_info_cache fci_cache;
-  zend_long nbytes;
-  bool nbytes_null;
+  zend_long nbytes, vcount, offset;
+  bool nbytes_null, vcount_null, offset_null;
   int fd; // php_socket_t fd;
   php_stream *stream;
-  size_t fnbytes;
+  size_t fnbytes, fvcount, foffset;
 
   obj = getThis();
   nbytes_null = true;
+  vcount_null = true;
+  offset_null = true;
   fci = empty_fcall_info;
   fci_cache = empty_fcall_info_cache;
   fd = -1;
 
-  ZEND_PARSE_PARAMETERS_START(3, 3)
+  ZEND_PARSE_PARAMETERS_START(5, 5)
   Z_PARAM_RESOURCE(res)
   Z_PARAM_LONG_OR_NULL(nbytes, nbytes_null)
+  Z_PARAM_LONG_OR_NULL(vcount, vcount_null)
+  Z_PARAM_LONG_OR_NULL(offset, offset_null)
   Z_PARAM_FUNC(fci, fci_cache)
   ZEND_PARSE_PARAMETERS_END();
 
@@ -552,6 +583,8 @@ static void php_mrloop_add_read_stream(INTERNAL_FUNCTION_PARAMETERS)
   PHP_STREAM_TO_FD(stream, res, fd);
 
   fnbytes = (size_t)(nbytes_null == true ? DEFAULT_STREAM_BUFF_LEN : nbytes);
+  fvcount = (size_t)(vcount_null == true ? DEFAULT_VECTOR_COUNT : vcount);
+  foffset = (size_t)(offset_null == true ? DEFAULT_READV_OFFSET : offset);
 
   iov = emalloc(sizeof(php_iovec_t));
   iov->iov_base = emalloc(fnbytes);
@@ -562,7 +595,7 @@ static void php_mrloop_add_read_stream(INTERNAL_FUNCTION_PARAMETERS)
 
   cb->data = iov;
 
-  mr_readvcb(this->loop, fd, iov, 1, 0, cb, php_mrloop_readv_cb);
+  mr_readvcb(this->loop, fd, iov, fvcount, foffset, cb, php_mrloop_readv_cb);
   mr_flush(this->loop);
 
   return;
@@ -576,18 +609,22 @@ static void php_mrloop_add_write_stream(INTERNAL_FUNCTION_PARAMETERS)
   php_iovec_t *iov;
   zend_fcall_info fci;
   zend_fcall_info_cache fci_cache;
+  zend_long vcount;
+  bool vcount_null;
   int fd;
   php_stream *stream;
-  size_t nbytes;
+  size_t nbytes, fvcount;
 
   obj = getThis();
   fci = empty_fcall_info;
   fci_cache = empty_fcall_info_cache;
   fd = -1;
+  vcount_null = true;
 
-  ZEND_PARSE_PARAMETERS_START(3, 3)
+  ZEND_PARSE_PARAMETERS_START(4, 4)
   Z_PARAM_RESOURCE(res)
   Z_PARAM_STR(contents)
+  Z_PARAM_LONG_OR_NULL(vcount, vcount_null)
   Z_PARAM_FUNC(fci, fci_cache)
   ZEND_PARSE_PARAMETERS_END();
 
@@ -607,7 +644,9 @@ static void php_mrloop_add_write_stream(INTERNAL_FUNCTION_PARAMETERS)
 
   cb->data = iov;
 
-  mr_writevcb(this->loop, fd, iov, 1, cb, php_mrloop_writev_cb);
+  fvcount = (size_t)(vcount_null == true ? DEFAULT_VECTOR_COUNT : vcount);
+
+  mr_writevcb(this->loop, fd, iov, fvcount, cb, php_mrloop_writev_cb);
   mr_flush(this->loop);
 
   return;
@@ -642,7 +681,7 @@ static void php_mrloop_writev(INTERNAL_FUNCTION_PARAMETERS)
 
     if (fcntl(fd, F_GETFD) < 0)
     {
-      PHP_MRLOOP_THROW("Detected invalid file descriptor");
+      PHP_MRLOOP_THROW(strerror(errno));
       mr_stop(this->loop);
 
       return;

src/loop.h

@@ -23,12 +23,15 @@
   ZEND_PARSE_PARAMETERS_END()
 #endif
 
-#define DEFAULT_STREAM_BUFF_LEN 8192
-#define DEFAULT_CONN_BUFF_LEN 65536
+#define DEFAULT_STREAM_BUFF_LEN 1024
+#define DEFAULT_CONN_BUFF_LEN 8132
 #define DEFAULT_HTTP_HEADER_LIMIT 100
+#define DEFAULT_VECTOR_COUNT 2
+#define DEFAULT_READV_OFFSET 0
 #define PHP_MRLOOP_TIMER 1
 #define PHP_MRLOOP_PERIODIC_TIMER 2
 #define PHP_MRLOOP_FUTURE_TICK 3
+#define PHP_MRLOOP_MAX_TCP_CONNECTIONS 1024
 
 struct php_mrloop_t;
 struct php_mrloop_cb_t;
@@ -60,7 +63,7 @@ struct php_mrloop_conn_t
   /* client socket address */
   char *addr;
   /* data sent over client socket */
-  char buffer[DEFAULT_CONN_BUFF_LEN];
+  char *buffer;
   /* client socket port */
   size_t port;
   /* scatter-gather I/O primitives */
@@ -97,6 +100,8 @@ php_mrloop_cb_t *tcp_cb;
 php_mrloop_cb_t *sig_cb[3];
 /* signal callback count */
 size_t sigc;
+/* TCP buffer size */
+size_t tcp_buff_size;
 ZEND_END_MODULE_GLOBALS(mrloop)
 /* }}} */
 

tests/002.phpt

@@ -10,6 +10,8 @@ $loop = Mrloop::init();
 $loop->addReadStream(
   $fd = \popen('pwd', 'r'),
   null,
+  null,
+  null,
   function (...$args) use ($fd, $loop) {
     [$message] = $args;