add port/socketpair

Author: ireader

include/port/socketpair.h

@@ -0,0 +1,69 @@
+#ifndef _socketpair_h_
+#define _socketpair_h_
+
+#if !defined(OS_WINDOWS)
+#include <sys/types.h> /* See NOTES */
+#include <sys/socket.h>
+
+// https://linux.die.net/man/3/socketpair
+#else
+
+#include "sockutil.h"
+#include <stdlib.h>
+#include <string.h>
+
+/// @param[in] domain AF_UNIX/AF_LOCAL only
+/// @param[in] type SOCK_STREAM/SOCK_DGRAM/SOCK_SEQPACKET only
+/// @param[in] protocol ignore, must be 0
+/// @return 0-ok, -1-error
+static inline int socketpair(int domain, int type, int protocol, socket_t sv[2])
+{
+	socket_t pr[2];
+	socklen_t addrlen;
+	struct addrinfo hints, *ai;
+	struct sockaddr_storage addr;
+
+	memset(&hints, 0, sizeof(hints));
+	hints.ai_family = protocol;
+	hints.ai_socktype = type;
+	hints.ai_flags = AI_PASSIVE;
+	if (0 != getaddrinfo(NULL, NULL, &hints, &ai))
+		return -1;
+
+	addrlen = ai->ai_addrlen;
+	memcpy(&addr, ai->ai_addr, ai->ai_addrlen);
+	freeaddrinfo(ai);
+
+	pr[0] = socket(domain, type, protocol);
+	pr[1] = socket(domain, type, protocol);
+	if (socket_invalid == pr[0] || socket_invalid == pr[1])
+		goto SOCKPAIRFAILED;
+
+	if(0 != socket_bind(pr[0], (struct sockaddr*)&addr, addrlen))
+		goto SOCKPAIRFAILED;
+
+	if (0 != socket_listen(pr[0], 1))
+		goto SOCKPAIRFAILED;
+
+	if(0 != socket_connect_by_time(pr[1], (struct sockaddr*)&addr, addrlen, 10))
+		goto SOCKPAIRFAILED;
+
+	sv[0] = socket_accept_by_time(pr[0], &addr, &addrlen, 10);
+	if(socket_invalid == sv[0])
+		goto SOCKPAIRFAILED;
+
+	socket_close(pr[0]); // close daemon socket
+	socket_setnonblock(sv[0], 1);
+	sv[1] = pr[1];
+	return 0;
+
+SOCKPAIRFAILED:
+	if (socket_invalid != pr[0])
+		socket_close(pr[0]);
+	if (socket_invalid != pr[1])
+		socket_close(pr[1]);
+	return -1;
+}
+#endif
+
+#endif /* !_socketpair_h_ */

libaio/include/aio-poll.h

@@ -23,15 +23,15 @@ int aio_poll_destroy(struct aio_poll_t* poll);
 /// @param[in] socket same as aio_poll_poll socket
 /// @param[in] flags AIO_POLL_IN/AIO_POLL_OUT
 /// @param[in] param same as aio_poll_poll param
-typedef void (*aio_poll_callback)(int code, socket_t socket, int flags, void* param);
+typedef void (*aio_poll_onpoll)(int code, socket_t socket, int flags, void* param);
 
 /// Async poll
 /// @param[in] flags AIO_POLL_IN/AIO_POLL_OUT
 /// @param[in] timeout poll timeout, in MS
 /// @param[in] callback poll callback
 /// @param[in] param user defined callback parameter
 /// @return 0-ok, other-error
-int aio_poll_poll(struct aio_poll_t* poll, socket_t socket, int flags, int timeout, aio_poll_callback callback, void* param);
+int aio_poll_poll(struct aio_poll_t* poll, socket_t socket, int flags, int timeout, aio_poll_onpoll callback, void* param);
 
 #ifdef __cplusplus
 }

libaio/src/aio-poll.c

@@ -3,13 +3,14 @@
 #include "sys/thread.h"
 #include "sys/locker.h"
 #include "sys/pollfd.h"
+#include "port/socketpair.h"
 #include "sockutil.h"
 #include "list.h"
 #include <errno.h>
 #include <stdlib.h>
 #include <assert.h>
 
-#define N_SOCKETS FD_SETSIZE
+#define N_SOCKETS (FD_SETSIZE-1)
 
 struct aio_poll_socket_t
 {
@@ -19,13 +20,14 @@ struct aio_poll_socket_t
 	int events;
 	int revents;
 	uint64_t expire;
-	aio_poll_callback callback;
+	aio_poll_onpoll callback;
 	void* param;
 };
 
 struct aio_poll_t
 {
 	locker_t locker;
+	socket_t pair[2];
 	struct list_head root;
 	struct list_head idles;
 
@@ -79,6 +81,13 @@ struct aio_poll_t* aio_poll_create()
 	{
 		LIST_INIT_HEAD(&poll->root);
 		LIST_INIT_HEAD(&poll->idles);
+		aio_poll_init_idles(poll);
+
+		if (0 != socketpair(PF_INET, SOCK_DGRAM, 0, poll->pair))
+		{
+			free(poll);
+			return NULL;
+		}
 
 		poll->running = 1;
 		locker_create(&poll->locker);
@@ -94,12 +103,16 @@ int aio_poll_destroy(struct aio_poll_t* poll)
 
 	assert(list_empty(&poll->root));
 	poll->running = 0;
+	socket_send_all_by_time(poll->pair[1], poll, 1, 0, 5000);
+
 	thread_destroy(poll->thread);
 	locker_destroy(&poll->locker);
+	socket_close(poll->pair[0]);
+	socket_close(poll->pair[1]);
 	return 0;
 }
 
-int aio_poll_poll(struct aio_poll_t* poll, socket_t socket, int flags, int timeout, aio_poll_callback callback, void* param)
+int aio_poll_poll(struct aio_poll_t* poll, socket_t socket, int flags, int timeout, aio_poll_onpoll callback, void* param)
 {
 	struct aio_poll_socket_t* s;
 
@@ -120,6 +133,9 @@ int aio_poll_poll(struct aio_poll_t* poll, socket_t socket, int flags, int timeo
 	locker_lock(&poll->locker);
 	list_insert_after(&s->link, poll->root.prev);
 	locker_unlock(&poll->locker);
+
+	// notify
+	socket_send_all_by_time(poll->pair[1], s, 1, 0, 5000);
 	return 0;
 }
 
@@ -128,28 +144,34 @@ static int STDCALL aio_poll_worker(void* param)
 {
 	int i, n, r;
 	uint64_t now;
+	char buf[128];
 	struct list_head* ptr;
 	struct aio_poll_t* poll;
-	struct aio_poll_socket_t* links[N_SOCKETS];
+	struct aio_poll_socket_t link0, *links[N_SOCKETS];
 
 	poll = (struct aio_poll_t*)param;
+	link0.fd = poll->pair[0];
+	link0.events = AIO_POLL_IN;
+	
 	while (poll->running)
 	{
+		socket_recv_by_time(poll->pair[0], buf, sizeof(buf), 0, 0);
+
 		n = 0;
+		links[n++] = &link0;
 		locker_lock(&poll->locker);
 		list_for_each(ptr, &poll->root)
 		{
 			links[n++] = list_entry(ptr, struct aio_poll_socket_t, link);
 		}
 		locker_unlock(&poll->locker);
 
-		// TODO: timeout
-		r = aio_poll_do(links, n, 1000);
+		r = aio_poll_do(links, n, 120000);
 		if (r < 0)
 			break;
 
 		now = system_clock();
-		for (i = 0; i < n; i++)
+		for (i = 1; i < n; i++)
 		{
 			if (0 != links[i]->revents)
 			{
@@ -231,11 +253,11 @@ static int aio_poll_do(struct aio_poll_socket_t* s[], int n, int timeout)
 
 	for (r = i = 0; i < n && i < 64; i++)
 	{
-		s[i]->oflags = 0;
+		s[i]->revents = 0;
 		if (fds[i].revents & POLLIN)
-			s[i]->oflags |= AIO_POLL_IN;
+			s[i]->revents |= AIO_POLL_IN;
 		if (fds[i].revents & POLLOUT)
-			s[i]->oflags |= AIO_POLL_OUT;
+			s[i]->revents |= AIO_POLL_OUT;
 	}
 
 	return r;