Merge branch 'master' of https://github.com/sogou/workflow into nossl

2026-10-25 00:43:28 +08:00 · 2024-03-05 21:11:41 +08:00
parent 0882a6c8d2 2baedff87c
commit ff08da425b
18 changed files with 613 additions and 97 deletions
@@ -71,8 +71,8 @@ static inline int __set_fd_nonblock(int fd)
 	return flags;
 }

-static int __bind_and_listen(int sockfd, const struct sockaddr *addr,
-							 socklen_t addrlen)
+static int __bind_sockaddr(int sockfd, const struct sockaddr *addr,
+						   socklen_t addrlen)
 {
 	struct sockaddr_storage ss;
 	socklen_t len;
@@ -94,7 +94,17 @@ static int __bind_and_listen(int sockfd, const struct sockaddr *addr,
 			return -1;
 	}

-	return listen(sockfd, SOMAXCONN < 4096 ? 4096 : SOMAXCONN);
+	return 0;
+}
+
+static void __release_conn(struct CommConnEntry *entry)
+{
+	delete entry->conn;
+	if (!entry->service)
+		pthread_mutex_destroy(&entry->mutex);
+
+	close(entry->sockfd);
+	free(entry);
 }

 int CommTarget::init(const struct sockaddr *addr, socklen_t addrlen,
@@ -132,6 +142,17 @@ void CommTarget::deinit()
 int CommMessageIn::feedback(const void *buf, size_t size)
 {
 	struct CommConnEntry *entry = this->entry;
+	CommSession *session = entry->session;
+	const struct sockaddr *addr;
+	socklen_t addrlen;
+	int ret;
+
+	if (session->passive && !session->reliable)
+	{
+		entry->target->get_addr(&addr, &addrlen);
+		return sendto(entry->sockfd, buf, size, 0, addr, addrlen);
+	}
+
 	return write(entry->sockfd, buf, size);
 }

@@ -266,10 +287,18 @@ CommSession::~CommSession()
 			entry = list_entry(pos, struct CommConnEntry, list);
 			list_del(pos);

-			errno_bak = errno;
-			mpoller_del(entry->sockfd, entry->mpoller);
-			entry->state = CONN_STATE_CLOSING;
-			errno = errno_bak;
+			if (this->reliable)
+			{
+				errno_bak = errno;
+				mpoller_del(entry->sockfd, entry->mpoller);
+				entry->state = CONN_STATE_CLOSING;
+				errno = errno_bak;
+			}
+			else
+			{
+				__release_conn(entry);
+				((CommServiceTarget *)target)->decref();
+			}
 		}

 		pthread_mutex_unlock(&target->mutex);
@@ -328,16 +357,6 @@ int Communicator::first_timeout_recv(CommSession *session)
 	return Communicator::first_timeout(session);
 }

-void Communicator::release_conn(struct CommConnEntry *entry)
-{
-	delete entry->conn;
-	if (!entry->service)
-		pthread_mutex_destroy(&entry->mutex);
-
-	close(entry->sockfd);
-	free(entry);
-}
-
 void Communicator::shutdown_service(CommService *service)
 {
 	close(service->listen_fd);
@@ -579,7 +598,7 @@ void Communicator::handle_incoming_request(struct poller_result *res)

 		if (__sync_sub_and_fetch(&entry->ref, 1) == 0)
 		{
-			this->release_conn(entry);
+			__release_conn(entry);
 			((CommServiceTarget *)target)->decref();
 		}
 	}
@@ -661,7 +680,7 @@ void Communicator::handle_incoming_reply(struct poller_result *res)
 		}

 		if (__sync_sub_and_fetch(&entry->ref, 1) == 0)
-			this->release_conn(entry);
+			__release_conn(entry);
 	}
 }

@@ -733,7 +752,7 @@ void Communicator::handle_reply_result(struct poller_result *res)
 		session->handle(state, res->error);
 		if (__sync_sub_and_fetch(&entry->ref, 1) == 0)
 		{
-			this->release_conn(entry);
+			__release_conn(entry);
 			((CommServiceTarget *)target)->decref();
 		}

@@ -786,7 +805,7 @@ void Communicator::handle_request_result(struct poller_result *res)
 		/* do nothing */
 		pthread_mutex_unlock(&entry->mutex);
 		if (__sync_sub_and_fetch(&entry->ref, 1) == 0)
-			this->release_conn(entry);
+			__release_conn(entry);

 		break;
 	}
@@ -819,7 +838,7 @@ struct CommConnEntry *Communicator::accept_conn(CommServiceTarget *target,
 			if (entry->conn)
 			{
 				entry->seq = 0;
-				entry->mpoller = this->mpoller;
+				entry->mpoller = NULL;
 				entry->service = service;
 				entry->target = target;
 				entry->sockfd = target->sockfd;
@@ -887,7 +906,7 @@ void Communicator::handle_connect_result(struct poller_result *res)

 		target->release();
 		session->handle(state, res->error);
-		this->release_conn(entry);
+		__release_conn(entry);
 		break;
 	}
 }
@@ -903,9 +922,10 @@ void Communicator::handle_listen_result(struct poller_result *res)
 	{
 	case PR_ST_SUCCESS:
 		target = (CommServiceTarget *)res->data.result;
-		entry = this->accept_conn(target, service);
+		entry = Communicator::accept_conn(target, service);
 		if (entry)
 		{
+			entry->mpoller = this->mpoller;
 			res->data.operation = PD_OP_READ;
 			res->data.fd = entry->sockfd;
 			res->data.create_message = Communicator::create_request;
@@ -919,7 +939,7 @@ void Communicator::handle_listen_result(struct poller_result *res)
 				break;
 			}

-			this->release_conn(entry);
+			__release_conn(entry);
 		}
 		else
 			close(target->sockfd);
@@ -938,6 +958,54 @@ void Communicator::handle_listen_result(struct poller_result *res)
 	}
 }

+void Communicator::handle_recvfrom_result(struct poller_result *res)
+{
+	CommService *service = (CommService *)res->data.context;
+	struct CommConnEntry *entry;
+	CommTarget *target;
+	int state, error;
+
+	switch (res->state)
+	{
+	case PR_ST_SUCCESS:
+		entry = (struct CommConnEntry *)res->data.result;
+		target = entry->target;
+		if (entry->state == CONN_STATE_SUCCESS)
+		{
+			state = CS_STATE_TOREPLY;
+			error = 0;
+			entry->state = CONN_STATE_IDLE;
+			list_add(&entry->list, &target->idle_list);
+		}
+		else
+		{
+			state = CS_STATE_ERROR;
+			if (entry->state == CONN_STATE_ERROR)
+				error = entry->error;
+			else
+				error = EBADMSG;
+		}
+
+		entry->session->handle(state, error);
+		if (state == CS_STATE_ERROR)
+		{
+			__release_conn(entry);
+			((CommServiceTarget *)target)->decref();
+		}
+
+		break;
+
+	case PR_ST_DELETED:
+		this->shutdown_service(service);
+		break;
+
+	case PR_ST_ERROR:
+	case PR_ST_STOPPED:
+		service->handle_stop(res->error);
+		break;
+	}
+}
+
 void Communicator::handle_sleep_result(struct poller_result *res)
 {
 	SleepSession *session = (SleepSession *)res->data.context;
@@ -1028,6 +1096,9 @@ void Communicator::handler_thread_routine(void *context)
 		case PD_OP_LISTEN:
 			comm->handle_listen_result(res);
 			break;
+		case PD_OP_RECVFROM:
+			comm->handle_recvfrom_result(res);
+			break;
 		case PD_OP_EVENT:
 		case PD_OP_NOTIFY:
 			comm->handle_aio_result(res);
@@ -1132,6 +1203,7 @@ poller_message_t *Communicator::create_request(void *context)
 		return NULL;

 	session->passive = 1;
+	session->reliable = 1;
 	entry->session = session;
 	session->target = target;
 	session->conn = entry->conn;
@@ -1184,6 +1256,59 @@ poller_message_t *Communicator::create_reply(void *context)
 	return session->in;
 }

+int Communicator::recv_request(const void *buf, size_t size,
+							   struct CommConnEntry *entry)
+{
+	CommService *service = entry->service;
+	CommTarget *target = entry->target;
+	CommSession *session;
+	size_t n;
+	int ret;
+
+	session = service->new_session(entry->seq, entry->conn);
+	if (!session)
+		return -1;
+
+	session->passive = 1;
+	session->reliable = 0;
+	entry->session = session;
+	session->target = target;
+	session->conn = entry->conn;
+	session->seq = entry->seq++;
+	session->out = NULL;
+	session->in = NULL;
+
+	entry->state = CONN_STATE_RECEIVING;
+
+	((CommServiceTarget *)target)->incref();
+
+	session->in = session->message_in();
+	if (session->in)
+	{
+		session->in->entry = entry;
+		do
+		{
+			n = size;
+			ret = session->in->append(buf, &n);
+			if (ret == 0)
+			{
+				size -= n;
+				buf = (const char *)buf + n;
+			}
+			else if (ret < 0)
+			{
+				entry->error = errno;
+				entry->state = CONN_STATE_ERROR;
+			}
+			else
+				entry->state = CONN_STATE_SUCCESS;
+
+		} while (ret == 0 && size > 0);
+	}
+
+	return 0;
+}
+
 int Communicator::partial_written(size_t n, void *context)
 {
 	struct CommConnEntry *entry = (struct CommConnEntry *)context;
@@ -1215,6 +1340,41 @@ void *Communicator::accept(const struct sockaddr *addr, socklen_t addrlen,
 		delete target;
 	}

+	close(sockfd);
+	return NULL;
+}
+
+void *Communicator::recvfrom(const struct sockaddr *addr, socklen_t addrlen,
+							 const void *buf, size_t size, void *context)
+{
+	CommService *service = (CommService *)context;
+	struct CommConnEntry *entry;
+	CommServiceTarget *target;
+	void *result;
+	int sockfd;
+
+	sockfd = dup(service->listen_fd);
+	if (sockfd >= 0)
+	{
+		result = Communicator::accept(addr, addrlen, sockfd, context);
+		if (result)
+		{
+			target = (CommServiceTarget *)result;
+			entry = Communicator::accept_conn(target, service);
+			if (entry)
+			{
+				if (Communicator::recv_request(buf, size, entry) >= 0)
+					return entry;
+
+				__release_conn(entry);
+			}
+			else
+				close(sockfd);
+
+			target->decref();
+		}
+	}
+
 	return NULL;
 }

@@ -1342,7 +1502,7 @@ struct CommConnEntry *Communicator::launch_conn(CommSession *session,
 	int sockfd;
 	int ret;

-	sockfd = this->nonblock_connect(target);
+	sockfd = Communicator::nonblock_connect(target);
 	if (sockfd >= 0)
 	{
 		entry = (struct CommConnEntry *)malloc(sizeof (struct CommConnEntry));
@@ -1355,7 +1515,7 @@ struct CommConnEntry *Communicator::launch_conn(CommSession *session,
 				if (entry->conn)
 				{
 					entry->seq = 0;
-					entry->mpoller = this->mpoller;
+					entry->mpoller = NULL;
 					entry->service = NULL;
 					entry->target = target;
 					entry->session = session;
@@ -1437,9 +1597,10 @@ int Communicator::request_new_conn(CommSession *session, CommTarget *target)
 	struct poller_data data;
 	int timeout;

-	entry = this->launch_conn(session, target);
+	entry = Communicator::launch_conn(session, target);
 	if (entry)
 	{
+		entry->mpoller = this->mpoller;
 		session->conn = entry->conn;
 		session->seq = entry->seq++;
 		data.operation = PD_OP_CONNECT;
@@ -1449,7 +1610,7 @@ int Communicator::request_new_conn(CommSession *session, CommTarget *target)
 		if (mpoller_add(&data, timeout, this->mpoller) >= 0)
 			return 0;

-		this->release_conn(entry);
+		__release_conn(entry);
 	}

 	return -1;
@@ -1483,18 +1644,24 @@ int Communicator::request(CommSession *session, CommTarget *target)
 	return 0;
 }

-int Communicator::nonblock_listen(CommService *service)
+int Communicator::nonblock_listen(CommService *service, int *reliable)
 {
 	int sockfd = service->create_listen_fd();
+	int ret;

 	if (sockfd >= 0)
 	{
 		if (__set_fd_nonblock(sockfd) >= 0)
 		{
-			if (__bind_and_listen(sockfd, service->bind_addr,
-								  service->addrlen) >= 0)
+			if (__bind_sockaddr(sockfd, service->bind_addr,
+								service->addrlen) >= 0)
 			{
-				return sockfd;
+				ret = listen(sockfd, SOMAXCONN);
+				if (ret >= 0 || errno == EOPNOTSUPP)
+				{
+					*reliable = (ret >= 0);
+					return sockfd;
+				}
 			}
 		}

@@ -1507,20 +1674,34 @@ int Communicator::nonblock_listen(CommService *service)
 int Communicator::bind(CommService *service)
 {
 	struct poller_data data;
+	int errno_bak = errno;
+	int reliable;
 	int sockfd;

-	sockfd = this->nonblock_listen(service);
+	sockfd = this->nonblock_listen(service, &reliable);
 	if (sockfd >= 0)
 	{
 		service->listen_fd = sockfd;
 		service->ref = 1;
-		data.operation = PD_OP_LISTEN;
 		data.fd = sockfd;
-		data.accept = Communicator::accept;
 		data.context = service;
 		data.result = NULL;
+		if (reliable)
+		{
+			data.operation = PD_OP_LISTEN;
+			data.accept = Communicator::accept;
+		}
+		else
+		{
+			data.operation = PD_OP_RECVFROM;
+			data.recvfrom = Communicator::recvfrom;
+		}
+
 		if (mpoller_add(&data, service->listen_timeout, this->mpoller) >= 0)
+		{
+			errno = errno_bak;
 			return 0;
+		}

 		close(sockfd);
 	}
@@ -1572,6 +1753,62 @@ int Communicator::reply_idle_conn(CommSession *session, CommTarget *target)
 	return ret;
 }

+int Communicator::reply_message_unreliable(struct CommConnEntry *entry)
+{
+	struct iovec vectors[ENCODE_IOV_MAX];
+	int cnt;
+
+	cnt = entry->session->out->encode(vectors, ENCODE_IOV_MAX);
+	if ((unsigned int)cnt > ENCODE_IOV_MAX)
+	{
+		if (cnt > ENCODE_IOV_MAX)
+			errno = EOVERFLOW;
+		return -1;
+	}
+
+	if (cnt > 0)
+	{
+		struct msghdr message = {
+			.msg_name		=	entry->target->addr,
+			.msg_namelen	=	entry->target->addrlen,
+			.msg_iov		=	vectors,
+			.msg_iovlen		=	(size_t)cnt,
+		};
+		if (sendmsg(entry->sockfd, &message, 0) < 0)
+			return -1;
+	}
+
+	return 0;
+}
+
+int Communicator::reply_unreliable(CommSession *session, CommTarget *target)
+{
+	struct CommConnEntry *entry;
+	struct list_head *pos;
+
+	if (!list_empty(&target->idle_list))
+	{
+		pos = target->idle_list.next;
+		entry = list_entry(pos, struct CommConnEntry, list);
+		list_del(pos);
+
+		target = entry->target;
+		session->out = session->message_out();
+		if (session->out)
+		{
+			if (this->reply_message_unreliable(entry) >= 0)
+				return 0;
+		}
+
+		__release_conn(entry);
+		((CommServiceTarget *)target)->decref();
+	}
+	else
+		errno = ENOENT;
+
+	return -1;
+}
+
 int Communicator::reply(CommSession *session)
 {
 	struct CommConnEntry *entry;
@@ -1588,9 +1825,10 @@ int Communicator::reply(CommSession *session)
 	errno_bak = errno;
 	session->passive = 2;
 	target = session->target;
-	ret = this->reply_idle_conn(session, target);
-	if (ret < 0)
-		return -1;
+	if (session->reliable)
+		ret = this->reply_idle_conn(session, target);
+	else
+		ret = this->reply_unreliable(session, target);

 	if (ret == 0)
 	{
@@ -1598,10 +1836,12 @@ int Communicator::reply(CommSession *session)
 		session->handle(CS_STATE_SUCCESS, 0);
 		if (__sync_sub_and_fetch(&entry->ref, 1) == 0)
 		{
-			this->release_conn(entry);
+			__release_conn(entry);
 			((CommServiceTarget *)target)->decref();
 		}
 	}
+	else if (ret < 0)
+		return -1;

 	errno = errno_bak;
 	return 0;
@@ -1623,7 +1863,13 @@ int Communicator::push(const void *buf, size_t size, CommSession *session)
 	if (!list_empty(&target->idle_list))
 	{
 		entry = list_entry(target->idle_list.next, struct CommConnEntry, list);
-		ret = write(entry->sockfd, buf, size);
+		if (session->reliable)
+			ret = write(entry->sockfd, buf, size);
+		else
+		{
+			ret = sendto(entry->sockfd, buf, size, 0,
+						 target->addr, target->addrlen);
+		}
 	}
 	else
 	{
@@ -1639,6 +1885,7 @@ int Communicator::shutdown(CommSession *session)
 {
 	CommTarget *target = session->target;
 	struct CommConnEntry *entry;
+	struct list_head *pos;
 	int ret;

 	if (session->passive != 1)
@@ -1651,10 +1898,21 @@ int Communicator::shutdown(CommSession *session)
 	pthread_mutex_lock(&target->mutex);
 	if (!list_empty(&target->idle_list))
 	{
-		entry = list_entry(target->idle_list.next, struct CommConnEntry, list);
-		list_del(&entry->list);
-		ret = mpoller_del(entry->sockfd, entry->mpoller);
-		entry->state = CONN_STATE_CLOSING;
+		pos = target->idle_list.next;
+		entry = list_entry(pos, struct CommConnEntry, list);
+		list_del(pos);
+
+		if (session->reliable)
+		{
+			ret = mpoller_del(entry->sockfd, entry->mpoller);
+			entry->state = CONN_STATE_CLOSING;
+		}
+		else
+		{
+			ret = 0;
+			__release_conn(entry);
+			((CommServiceTarget *)target)->decref();
+		}
 	}
 	else
 	{
@@ -30,9 +30,8 @@

 class CommConnection
 {
-protected:
+public:
 	virtual ~CommConnection() { }
-	friend class Communicator;
 };

 class CommTarget
@@ -144,7 +143,8 @@ private:
 private:
 	struct timespec begin_time;
 	int timeout;
-	int passive;
+	short passive;
+	short reliable;

 public:
 	CommSession() { this->passive = 0; }
@@ -271,16 +271,6 @@ private:

 	int create_handler_threads(size_t handler_threads);

-	int nonblock_connect(CommTarget *target);
-	int nonblock_listen(CommService *service);
-
-	struct CommConnEntry *launch_conn(CommSession *session,
-									  CommTarget *target);
-	struct CommConnEntry *accept_conn(class CommServiceTarget *target,
-									  CommService *service);
-
-	void release_conn(struct CommConnEntry *entry);
-
 	void shutdown_service(CommService *service);

 	void shutdown_io_service(IOService *service);
@@ -292,10 +282,14 @@ private:

 	int send_message(struct CommConnEntry *entry);

+	int request_new_conn(CommSession *session, CommTarget *target);
+
 	int request_idle_conn(CommSession *session, CommTarget *target);
 	int reply_idle_conn(CommSession *session, CommTarget *target);

-	int request_new_conn(CommSession *session, CommTarget *target);
+	int reply_message_unreliable(struct CommConnEntry *entry);
+
+	int reply_unreliable(CommSession *session, CommTarget *target);

 	void handle_incoming_request(struct poller_result *res);
 	void handle_incoming_reply(struct poller_result *res);
@@ -309,12 +303,22 @@ private:
 	void handle_connect_result(struct poller_result *res);
 	void handle_listen_result(struct poller_result *res);

+	void handle_recvfrom_result(struct poller_result *res);
+
 	void handle_sleep_result(struct poller_result *res);

 	void handle_aio_result(struct poller_result *res);

 	static void handler_thread_routine(void *context);

+	static int nonblock_connect(CommTarget *target);
+	static int nonblock_listen(CommService *service, int *reliable);
+
+	static struct CommConnEntry *launch_conn(CommSession *session,
+											 CommTarget *target);
+	static struct CommConnEntry *accept_conn(class CommServiceTarget *target,
+											 CommService *service);
+
 	static int first_timeout(CommSession *session);
 	static int next_timeout(CommSession *session);

@@ -329,11 +333,17 @@ private:
 	static poller_message_t *create_request(void *context);
 	static poller_message_t *create_reply(void *context);

+	static int recv_request(const void *buf, size_t size,
+							struct CommConnEntry *entry);
+
 	static int partial_written(size_t n, void *context);

 	static void *accept(const struct sockaddr *addr, socklen_t addrlen,
 						int sockfd, void *context);

+	static void *recvfrom(const struct sockaddr *addr, socklen_t addrlen,
+						  const void *buf, size_t size, void *context);
+
 	static void callback(struct poller_result *res, void *context);

 public:
@@ -535,10 +535,7 @@ static void __poller_handle_listen(struct __poller_node *node,

 		result = node->data.accept(addr, addrlen, sockfd, node->data.context);
 		if (!result)
-		{
-			close(sockfd);
 			break;
-		}

 		res->data = node->data;
 		res->data.result = result;