Merge pull request #59 from Barenboim/channel

merge from Barenboim/workflow-dev channel

CMakeLists_Headers.txt

@@ -7,6 +7,7 @@ set(COMMON_KERNEL_HEADERS
 	src/kernel/SleepRequest.h
 	src/kernel/ExecRequest.h
 	src/kernel/IORequest.h
+	src/kernel/TransRequest.h
 	src/kernel/Executor.h
 	src/kernel/list.h
 	src/kernel/mpoller.h

src/kernel/CMakeLists.txt

@@ -17,6 +17,7 @@ set(SRC
 	msgqueue.c
 	thrdpool.c
 	CommRequest.cc
+	TransRequest.cc
 	CommScheduler.cc
 	Communicator.cc
 	Executor.cc

src/kernel/CommScheduler.h

@@ -171,6 +171,33 @@ public:
 		this->comm.io_unbind(service);
 	}
 
+	/* for channels. */
+	int establish(CommChannel *channel, CommSchedObject *object,
+				  int wait_timeout, CommTarget **target)
+	{
+		int ret = -1;
+
+		*target = object->acquire(wait_timeout);
+		if (*target)
+		{
+			ret = this->comm.establish(channel, *target);
+			if (ret < 0)
+				(*target)->release();
+		}
+
+		return ret;
+	}
+
+	int send(TransSession *session, CommChannel *channel)
+	{
+		return this->comm.send(session, channel);
+	}
+
+	void shutdown(CommChannel *channel)
+	{
+		this->comm.shutdown(channel);
+	}
+
 public:
 	int is_handler_thread() const
 	{

src/kernel/Communicator.cc

@@ -24,7 +24,6 @@
 #include <time.h>
 #include <fcntl.h>
 #include <unistd.h>
-#include <stddef.h>
 #include <stdlib.h>
 #include <string.h>
 #include <pthread.h>
@@ -43,6 +42,7 @@ struct CommConnEntry
 	CommConnection *conn;
 	long long seq;
 	int sockfd;
+	short channel_flag;
 #define CONN_STATE_CONNECTING	0
 #define CONN_STATE_CONNECTED	1
 #define CONN_STATE_RECEIVING	2
@@ -51,7 +51,7 @@ struct CommConnEntry
 #define CONN_STATE_KEEPALIVE	5
 #define CONN_STATE_CLOSING		6
 #define CONN_STATE_ERROR		7
-	int state;
+	short state;
 	int error;
 	int ref;
 	struct iovec *write_iov;
@@ -119,6 +119,45 @@ static int __create_ssl(SSL_CTX *ssl_ctx, struct CommConnEntry *entry)
 	return -1;
 }
 
+#ifndef IOV_MAX
+# ifdef UIO_MAXIOV
+#  define IOV_MAX	UIO_MAXIOV
+# else
+#  define IOV_MAX	1024
+# endif
+#endif
+
+static int __send_vectors(struct iovec vectors[], int cnt,
+						  struct CommConnEntry *entry)
+{
+	ssize_t n;
+	int i;
+
+	while (cnt > 0)
+	{
+		n = writev(entry->sockfd, vectors, cnt <= IOV_MAX ? cnt : IOV_MAX);
+		if (n < 0)
+			return errno == EAGAIN ? cnt : -1;
+
+		for (i = 0; i < cnt; i++)
+		{
+			if ((size_t)n >= vectors[i].iov_len)
+				n -= vectors[i].iov_len;
+			else
+			{
+				vectors[i].iov_base = (char *)vectors[i].iov_base + n;
+				vectors[i].iov_len -= n;
+				break;
+			}
+		}
+
+		vectors += i;
+		cnt -= i;
+	}
+
+	return 0;
+}
+
 int CommTarget::init(const struct sockaddr *addr, socklen_t addrlen,
 					 int connect_timeout, int response_timeout)
 {
@@ -304,6 +343,21 @@ CommSession::~CommSession()
 	((CommServiceTarget *)target)->decref();
 }
 
+class CommMessageOutEmpty : public CommMessageOut
+{
+private:
+	virtual int encode(struct iovec *vectors, int max)
+	{
+		return 0;
+	}
+};
+
+CommMessageOut *CommChannel::message_out()
+{
+	static CommMessageOutEmpty empty;
+	return &empty;
+}
+
 inline int Communicator::first_timeout(CommSession *session)
 {
 	int timeout = session->target->response_timeout;
@@ -389,30 +443,10 @@ int Communicator::send_message_sync(struct iovec vectors[], int cnt,
 	CommSession *session = entry->session;
 	CommService *service;
 	int timeout;
-	ssize_t n;
-	int i;
 
-	while (cnt > 0)
-	{
-		n = writev(entry->sockfd, vectors, cnt <= IOV_MAX ? cnt : IOV_MAX);
-		if (n < 0)
-			return errno == EAGAIN ? cnt : -1;
-
-		for (i = 0; i < cnt; i++)
-		{
-			if ((size_t)n >= vectors[i].iov_len)
-				n -= vectors[i].iov_len;
-			else
-			{
-				vectors[i].iov_base = (char *)vectors[i].iov_base + n;
-				vectors[i].iov_len -= n;
-				break;
-			}
-		}
-
-		vectors += i;
-		cnt -= i;
-	}
+	cnt = __send_vectors(vectors, cnt, entry);
+	if (cnt != 0)
+		return cnt;
 
 	service = entry->service;
 	if (service)
@@ -529,7 +563,7 @@ int Communicator::send_message(struct CommConnEntry *entry)
 	}
 
 	end = vectors + cnt;
-	if (!entry->ssl)
+	if (!entry->ssl || cnt == 0)
 	{
 		cnt = this->send_message_sync(vectors, cnt, entry);
 		if (cnt <= 0)
@@ -837,12 +871,10 @@ struct CommConnEntry *Communicator::accept_conn(CommServiceTarget *target,
 												CommService *service)
 {
 	struct CommConnEntry *entry;
-	size_t size;
 
 	if (__set_fd_nonblock(target->sockfd) >= 0)
 	{
-		size = offsetof(struct CommConnEntry, mutex);
-		entry = (struct CommConnEntry *)malloc(size);
+		entry = (struct CommConnEntry *)malloc(sizeof (struct CommConnEntry));
 		if (entry)
 		{
 			entry->conn = service->new_connection(target->sockfd);
@@ -855,6 +887,7 @@ struct CommConnEntry *Communicator::accept_conn(CommServiceTarget *target,
 				entry->ssl = NULL;
 				entry->sockfd = target->sockfd;
 				entry->state = CONN_STATE_CONNECTED;
+				entry->channel_flag = 0;
 				entry->ref = 1;
 				return entry;
 			}
@@ -996,7 +1029,9 @@ void Communicator::handle_connect_result(struct poller_result *res)
 
 		target->release();
 		session->handle(state, res->error);
-		this->release_conn(entry);
+		if (__sync_sub_and_fetch(&entry->ref, 1) == 0)
+			this->release_conn(entry);
+
 		break;
 	}
 }
@@ -1257,7 +1292,117 @@ int Communicator::partial_written(size_t n, void *context)
 void Communicator::callback(struct poller_result *res, void *context)
 {
 	Communicator *comm = (Communicator *)context;
-	msgqueue_put(res, comm->queue);
+	struct CommConnEntry *entry;
+	CommSession *session;
+	int state;
+
+	switch (res->data.operation)
+	{
+	case PD_OP_READ:
+	case PD_OP_WRITE:
+	case PD_OP_CONNECT:
+	case PD_OP_SSL_CONNECT:
+		entry = (struct CommConnEntry *)res->data.context;
+		if (entry->channel_flag)
+			break;
+
+	default:
+		msgqueue_put(res, comm->queue);
+		return;
+	}
+
+	session = entry->session;
+	if (entry->state == CONN_STATE_CONNECTING)
+	{
+		comm->handle_connect_result(res);
+		if (entry->ref == 1)
+			comm->release_conn(entry);
+		else if (entry->state == CONN_STATE_RECEIVING)
+			((CommChannel *)session)->handle_established();
+	}
+	else if (entry->state == CONN_STATE_SUCCESS)
+	{
+		((CommChannel *)session)->handle_in(session->in);
+		session->in = NULL;
+		if (session->timeout == 0)
+			entry->state = CONN_STATE_CLOSING;
+		else
+		{
+			session->timeout = -1;
+			session->begin_time.tv_nsec = -1;
+			entry->state = CONN_STATE_RECEIVING;
+		}
+	}
+	else
+	{
+		if (res->data.operation == PD_OP_WRITE)
+		{
+			close(entry->sockfd);
+			free(entry->write_iov);
+			switch (res->state)
+			{
+			case PR_ST_FINISHED:
+				state = CS_STATE_SUCCESS;
+				break;
+			case PR_ST_ERROR:
+				state = CS_STATE_ERROR;
+				break;
+			case PR_ST_DELETED:
+			case PR_ST_STOPPED:
+				state = CS_STATE_STOPPED;
+				break;
+			}
+
+			free(entry);
+			entry = ((TransSession *)session)->channel->entry;
+			session->handle(state, res->error);
+			session = entry->session;
+		}
+		else
+		{
+			switch (res->state)
+			{
+			case PR_ST_FINISHED:
+				if (session->in)
+				{
+					res->error = ECONNRESET;
+			case PR_ST_ERROR:
+					state = CS_STATE_ERROR;
+				}
+				else
+					state = CS_STATE_SUCCESS;
+
+				break;
+
+			case PR_ST_DELETED:
+				if (entry->state == CONN_STATE_CLOSING)
+					state = CS_STATE_SUCCESS;
+				else
+			case PR_ST_STOPPED:
+					state = CS_STATE_STOPPED;
+
+				break;
+			}
+
+			if (entry->channel_flag == 1)
+			{
+				((CommChannel *)session)->handle_terminated();
+				entry->state = state;
+				entry->error = res->error;
+			}
+			else
+				state = CS_STATE_SHUTDOWN;
+		}
+
+		if (__sync_sub_and_fetch(&entry->ref, 1) == 0)
+		{
+			entry->target->release();
+			session->handle(state, res->error);
+			comm->release_conn(entry);
+		}
+	}
+
+	free(res);
 }
 
 void *Communicator::accept(const struct sockaddr *addr, socklen_t addrlen,
@@ -1421,6 +1566,7 @@ struct CommConnEntry *Communicator::launch_conn(CommSession *session,
 					entry->ssl = NULL;
 					entry->sockfd = sockfd;
 					entry->state = CONN_STATE_CONNECTING;
+					entry->channel_flag = 0;
 					entry->ref = 1;
 					return entry;
 				}
@@ -1657,6 +1803,111 @@ int Communicator::reply(CommSession *session)
 	return 0;
 }
 
+int Communicator::establish(CommChannel *channel, CommTarget *target)
+{
+	struct CommConnEntry *entry;
+	struct poller_data data;
+
+	entry = this->launch_conn(channel, target);
+	if (entry)
+	{
+		entry->channel_flag = 1;
+		entry->ref++;
+
+		channel->target = target;
+		channel->entry = entry;
+		channel->out = NULL;
+		channel->in = NULL;
+		channel->seq = 0;
+
+		data.operation = PD_OP_CONNECT;
+		data.fd = entry->sockfd;
+		data.ssl = NULL;
+		data.context = entry;
+		if (mpoller_add(&data, target->connect_timeout, this->mpoller) >= 0)
+			return 0;
+
+		this->release_conn(entry);
+	}
+
+	return -1;
+}
+
+int Communicator::send(TransSession *session, CommChannel *channel)
+{
+	struct CommConnEntry *entry = channel->entry;
+	struct iovec vectors[ENCODE_IOV_MAX];
+	struct iovec *end;
+	int cnt;
+
+	session->channel = channel;
+	session->target = channel->target;
+	session->seq = 0;
+	session->in = NULL;
+	session->out = session->message_out();
+	if (!session->out)
+		return -1;
+
+	cnt = session->out->encode(vectors, ENCODE_IOV_MAX);
+	if ((unsigned int)cnt > ENCODE_IOV_MAX)
+	{
+		if (cnt > ENCODE_IOV_MAX)
+			errno = EOVERFLOW;
+		return -1;
+	}
+
+	end = vectors + cnt;
+	if (!entry->ssl || cnt == 0)
+	{
+		cnt = __send_vectors(vectors, cnt, entry);
+		if (cnt == 0)
+			return 1;
+		else if (cnt < 0)
+			return -1;
+	}
+
+	entry = (struct CommConnEntry *)malloc(sizeof (struct CommConnEntry));
+	if (entry)
+	{
+		entry->sockfd = dup(channel->entry->sockfd);
+		if (entry->sockfd >= 0)
+		{
+			entry->session = session;
+			entry->ssl = channel->entry->ssl;
+			entry->state = CONN_STATE_KEEPALIVE;
+			entry->channel_flag = 1;
+			if (this->send_message_async(end - cnt, cnt, entry) >= 0)
+			{
+				__sync_add_and_fetch(&channel->entry->ref, 1);
+				return 0;
+			}
+
+			close(entry->sockfd);
+		}
+
+		free(entry);
+	}
+
+	return -1;
+}
+
+void Communicator::shutdown(CommChannel *channel)
+{
+	struct CommConnEntry *entry = channel->entry;
+	int errno_bak = errno;
+
+	entry->channel_flag = 2;
+	mpoller_del(entry->sockfd, this->mpoller);
+	if (__sync_sub_and_fetch(&entry->ref, 1) == 0)
+	{
+		entry->target->release();
+		channel->handle(entry->state, entry->error);
+		this->release_conn(entry);
+	}
+
+	errno = errno_bak;
+}
+
 int Communicator::sleep(SleepSession *session)
 {
 	struct timespec value;

src/kernel/Communicator.h

@@ -22,6 +22,7 @@
 #include <sys/types.h>
 #include <sys/socket.h>
 #include <sys/uio.h>
+#include <errno.h>
 #include <time.h>
 #include <stddef.h>
 #include <pthread.h>
@@ -124,6 +125,7 @@ public:
 #define CS_STATE_ERROR		1
 #define CS_STATE_STOPPED	2
 #define CS_STATE_TOREPLY	3	/* for service session only. */
+#define CS_STATE_SHUTDOWN	4	/* for channel only */
 
 class CommSession
 {
@@ -238,6 +240,49 @@ public:
 	friend class Communicator;
 };
 
+class CommChannel : public CommSession
+{
+private:
+	virtual CommMessageIn *message_in() = 0;
+	virtual int keep_alive_timeout() { return -1; }
+	virtual int first_timeout() { return -1; }
+	virtual void handle_established() = 0;
+	virtual void handle_in(CommMessageIn *msg) = 0;
+	virtual void handle_terminated() { }
+	virtual void handle(int state, int error) = 0;
+
+private:
+	virtual CommMessageOut *message_out(); /* final */
+	CommMessageOut *get_message_out() { return NULL; } /* deleted */
+
+private:
+	struct CommConnEntry *entry;
+	friend class Communicator;
+};
+
+class TransSession : public CommSession
+{
+private:
+	virtual CommMessageOut *message_out()
+	{
+		errno = ENOSYS;
+		return NULL;
+	}
+
+	virtual CommMessageIn *message_in()
+	{
+		errno = ENOSYS;
+		return NULL;
+	}
+
+protected:
+	CommChannel *get_channel() const { return this->channel; }
+
+private:
+	CommChannel *channel;
+	friend class Communicator;
+};
+
 #define SS_STATE_COMPLETE	0
 #define SS_STATE_ERROR		1
 #define SS_STATE_DISRUPTED	2
@@ -271,6 +316,10 @@ public:
 	int bind(CommService *service);
 	void unbind(CommService *service);
 
+	int establish(CommChannel *channel, CommTarget *target);
+	int send(TransSession *session, CommChannel *channel);
+	void shutdown(CommChannel *channel);
+
 	int sleep(SleepSession *session);
 
 	int io_bind(IOService *service);

src/kernel/TransRequest.cc

@@ -0,0 +1,59 @@
+/*
+  Copyright (c) 2021 Sogou, Inc.
+
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+
+  Author: Xie Han (xiehan@sogou-inc.com)
+*/
+
+#include <errno.h>
+#include "CommScheduler.h"
+#include "TransRequest.h"
+
+void ChanRequest::dispatch()
+{
+	if (!this->established)
+	{
+		if (this->scheduler->establish(this, this->object, this->wait_timeout,
+									   &this->target) < 0)
+		{
+			this->state = CS_STATE_ERROR;
+			this->error = errno;
+			this->subtask_done();
+		}
+	}
+	else
+		this->scheduler->shutdown(this);
+}
+
+void TransRequest::dispatch()
+{
+	int ret = this->scheduler->send(this, this->channel);
+
+	if (ret == 0)
+		return;
+
+	if (ret > 0)
+	{
+		this->state = CS_STATE_SUCCESS;
+		this->error = 0;
+	}
+	else
+	{
+		this->state = CS_STATE_ERROR;
+		this->error = errno;
+	}
+
+	this->subtask_done();
+}
+

src/kernel/TransRequest.h

@@ -0,0 +1,109 @@
+/*
+  Copyright (c) 2021 Sogou, Inc.
+
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+
+  Author: Xie Han (xiehan@sogou-inc.com)
+*/
+
+#ifndef _TRANSREQUEST_H_
+#define _TRANSREQUEST_H_
+
+#include "SubTask.h"
+#include "Communicator.h"
+#include "CommScheduler.h"
+
+class ChanRequest : public SubTask, public CommChannel
+{
+public:
+	ChanRequest(CommSchedObject *object, CommScheduler *scheduler)
+	{
+		this->scheduler = scheduler;
+		this->object = object;
+		this->wait_timeout = 0;
+		this->established = 0;
+	}
+
+	CommSchedObject *get_request_object() const { return this->object; }
+	void set_request_object(CommSchedObject *object) { this->object = object; }
+	int get_wait_timeout() const { return this->wait_timeout; }
+	void set_wait_timeout(int timeout) { this->wait_timeout = timeout; }
+
+public:
+	virtual void dispatch();
+
+protected:
+	int state;
+	int error;
+
+protected:
+	CommTarget *target;
+
+protected:
+	int established;
+	int wait_timeout;
+	CommSchedObject *object;
+	CommScheduler *scheduler;
+
+protected:
+	virtual void handle_established()
+	{
+		this->state = CS_STATE_SUCCESS;
+		this->error = 0;
+		this->established = 1;
+		this->subtask_done();
+	}
+
+	virtual void handle(int state, int error)
+	{
+		this->state = state;
+		this->error = error;
+		this->established = 0;
+		this->subtask_done();
+	}
+};
+
+class TransRequest : public SubTask, public TransSession
+{
+public:
+	TransRequest(CommChannel *channel, CommScheduler *scheduler)
+	{
+		this->scheduler = scheduler;
+		this->channel = channel;
+	}
+
+	CommChannel *get_request_channel() const { return this->channel; }
+	void set_request_channel(CommChannel *channel) { this->channel = channel; }
+
+public:
+	virtual void dispatch();
+
+protected:
+	int state;
+	int error;
+
+protected:
+	CommChannel *channel;
+	CommScheduler *scheduler;
+
+protected:
+	virtual void handle(int state, int error)
+	{
+		this->state = state;
+		this->error = error;
+		this->subtask_done();
+	}
+};
+
+#endif
+