Replace thread engine with job engine

1 files changed, 99 insertions, 66 deletions

diff --git a/src/threadengines/threadengine_pthread.cpp b/src/threadengines/threadengine_pthread.cpp
index 29a4b1a96..5844ae3d7 100644
--- a/src/threadengines/threadengine_pthread.cpp
+++ b/src/threadengines/threadengine_pthread.cpp
@@ -17,62 +17,23 @@
 #include <signal.h>
 #include <fcntl.h>
 
-ThreadEngine::ThreadEngine()
-{
-}
-
-static void* entry_point(void* parameter)
-{
-	/* Recommended by nenolod, signal safety on a per-thread basis */
-	sigset_t set;
-	sigemptyset(&set);
-	sigaddset(&set, SIGPIPE);
-	pthread_sigmask(SIG_BLOCK, &set, NULL);
-
-	Thread* pt = static_cast<Thread*>(parameter);
-	pt->Run();
-	return parameter;
-}
-
-
-void ThreadEngine::Start(Thread* thread)
-{
-	ThreadData* data = new ThreadData;
-	thread->state = data;
-
-	if (pthread_create(&data->pthread_id, NULL, entry_point, thread) != 0)
-	{
-		thread->state = NULL;
-		delete data;
-		throw CoreException("Unable to create new thread: " + std::string(strerror(errno)));
-	}
-}
-
-ThreadEngine::~ThreadEngine()
-{
-}
-
-void ThreadData::FreeThread(Thread* thread)
-{
-	thread->SetExitFlag();
-	pthread_join(pthread_id, NULL);
-}
-
 #ifdef HAS_EVENTFD
 #include <sys/eventfd.h>
 
 class ThreadSignalSocket : public EventHandler
 {
-	SocketThread* parent;
  public:
-	ThreadSignalSocket(SocketThread* p, int newfd) : parent(p)
+	ThreadSignalSocket()
 	{
-		SetFd(newfd);
+		SetFd(eventfd(0, EFD_NONBLOCK));
+		if (fd < 0)
+			throw CoreException("Could not create eventfd " + std::string(strerror(errno)));
 		ServerInstance->SE->AddFd(this, FD_WANT_FAST_READ | FD_WANT_NO_WRITE);
 	}
 
 	~ThreadSignalSocket()
 	{
+		close(fd);
 	}
 
 	void Notify()
@@ -86,33 +47,32 @@ class ThreadSignalSocket : public EventHandler
 		{
 			eventfd_t dummy;
 			eventfd_read(fd, &dummy);
-			parent->OnNotify();
+			ServerInstance->Threads->result_loop();
 		}
 		else
 		{
+			ServerInstance->Threads->job_lock.lock();
+			ServerInstance->Threads->result_s = NULL;
+			ServerInstance->Threads->job_lock.unlock();
 			ServerInstance->GlobalCulls.AddItem(this);
 		}
 	}
 };
 
-SocketThread::SocketThread()
-{
-	int fd = eventfd(0, EFD_NONBLOCK);
-	if (fd < 0)
-		throw new CoreException("Could not create pipe " + std::string(strerror(errno)));
-	signal.sock = new ThreadSignalSocket(this, fd);
-}
 #else
 
 class ThreadSignalSocket : public EventHandler
 {
-	SocketThread* parent;
 	int send_fd;
  public:
-	ThreadSignalSocket(SocketThread* p, int recvfd, int sendfd) :
-		parent(p), send_fd(sendfd)
+	ThreadSignalSocket()
 	{
-		SetFd(recvfd);
+		int fds[2];
+		if (pipe(fds))
+			throw new CoreException("Could not create pipe " + std::string(strerror(errno)));
+		SetFd(fds[0]);
+		send_fd = fds[1];
+
 		ServerInstance->SE->NonBlocking(fd);
 		ServerInstance->SE->AddFd(this, FD_WANT_FAST_READ | FD_WANT_NO_WRITE);
 	}
@@ -120,6 +80,7 @@ class ThreadSignalSocket : public EventHandler
 	~ThreadSignalSocket()
 	{
 		close(send_fd);
+		close(fd);
 	}
 
 	void Notify()
@@ -134,29 +95,101 @@ class ThreadSignalSocket : public EventHandler
 		{
 			char dummy[128];
 			read(fd, dummy, 128);
-			parent->OnNotify();
+			ServerInstance->Threads->result_loop();
 		}
 		else
 		{
+			ServerInstance->Threads->job_lock.lock();
+			ServerInstance->Threads->result_s = NULL;
+			ServerInstance->Threads->job_lock.unlock();
 			ServerInstance->GlobalCulls.AddItem(this);
 		}
 	}
 };
+#endif
 
-SocketThread::SocketThread()
+ThreadEngine::ThreadEngine() : result_s(NULL)
 {
-	int fds[2];
-	if (pipe(fds))
-		throw new CoreException("Could not create pipe " + std::string(strerror(errno)));
-	signal.sock = new ThreadSignalSocket(this, fds[0], fds[1]);
 }
-#endif
 
-void SocketThread::NotifyParent()
+ThreadEngine::~ThreadEngine()
 {
-	signal.sock->Notify();
+	delete result_s;
 }
 
-SocketThread::~SocketThread()
+void* ThreadEngine::Runner::entry_point(void* parameter)
 {
+	/* Recommended by nenolod, signal safety on a per-thread basis */
+	sigset_t set;
+	sigemptyset(&set);
+	sigaddset(&set, SIGPIPE);
+	pthread_sigmask(SIG_BLOCK, &set, NULL);
+
+	Runner* te = static_cast<Runner*>(parameter);
+	te->main_loop();
+	return parameter;
+}
+
+ThreadEngine::Runner::Runner(ThreadEngine* t)
+	: te(t), current(NULL)
+{
+	if (pthread_create(&id, NULL, entry_point, this) != 0)
+		throw CoreException("Unable to create new thread: " + std::string(strerror(errno)));
+}
+
+ThreadEngine::Runner::~Runner()
+{
+	pthread_join(id, NULL);
+}
+
+void ThreadEngine::Submit(Job* job)
+{
+	Mutex::Lock lock(job_lock);
+	if (threads.empty())
+	{
+		Runner* thread = new Runner(this);
+		threads.push_back(thread);
+	}
+	if (result_s == NULL)
+		result_s = new ThreadSignalSocket();
+	submit_q.push_back(job);
+	submit_s.signal_one();
+}
+
+void ThreadEngine::Runner::main_loop()
+{
+	te->job_lock.lock();
+	while (1)
+	{
+		while (te->submit_q.empty())
+			te->submit_s.wait(te->job_lock);
+
+		current = te->submit_q.front();
+		te->submit_q.pop_front();
+		te->job_lock.unlock();
+		
+		current->run();
+
+		te->job_lock.lock();
+		te->result_q.push_back(current);
+		current = NULL;
+
+		if (te->result_s)
+			te->result_s->Notify();
+	}
+	te->job_lock.unlock();
+}
+
+void ThreadEngine::result_loop()
+{
+	job_lock.lock();
+	while (!result_q.empty())
+	{
+		Job* job = result_q.front();
+		result_q.pop_front();
+		job_lock.unlock();
+		job->finish();
+		job_lock.lock();
+	}
+	job_lock.unlock();
 }