Added UNIX socket and test for it

client/Socket.php

@@ -1,43 +1,15 @@
 <?php
 
-class Socket
+abstract class Socket
 {
 	const RECEIVE_TIMEOUT = 10;
 
-	private $sock;
+	protected $sock;
-	private $throw_exception_on_timeout;
+	protected $throw_exception_on_timeout;
-	private $connected;
+	protected $connected;
-	private $host;
-	private $port;
-
-	public function __construct($ip='', $port='', $throw_exception_on_timeout = false, $auto_connect = true)
-	{
-		$this->sock = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
-		if ($this->sock === FALSE) 
-		{
-			throw new Exception("Error creating socket: ".socket_last_error()." ".socket_strerror(socket_last_error()));
-		}
-		$this->connected = false;
-		$this->host = $ip;
-		$this->port = $port;
-		if ($auto_connect)
-		{
-			$this->connect();
-		}
-		$this->throw_exception_on_timeout = $throw_exception_on_timeout;
-	}
-
-	public function connect()
-	{
-		if ($this->connected) return;
-		// тушим ошибки, иначе вывалится с PHP Error вместо exception
-		if (@socket_connect($this->sock, $this->host, $this->port) === FALSE) 
-		{
-			throw new Exception("Error connecting: ".socket_last_error()." ".socket_strerror(socket_last_error()));
-		}
-		$this->connected = true;
-	}
 
+	abstract public function connect();
+	
 	/**
 	 * Закрытие соединения
 	 */
@@ -51,7 +23,7 @@ class Socket
 	}
 
 	/*
-	 * Пропихивает строку в сокет. Бросит exception если случилась лажа.
+	 * Пропихивает строку в сокет.
 	 */
 
 	public function iterativeSend($s)
@@ -156,4 +128,70 @@ class Socket
 		return $this->iterativeReceive($len, 1);
 	}
 
+}
+
+class IPSocket extends Socket
+{
+	private $host;
+	private $port;
+	
+	public function __construct($ip='', $port='', $throw_exception_on_timeout = false, $auto_connect = true)
+	{
+		$this->sock = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
+		if ($this->sock === FALSE) 
+		{
+			throw new Exception("Error creating socket: ".socket_last_error()." ".socket_strerror(socket_last_error()));
+		}
+		$this->connected = false;
+		$this->host = $ip;
+		$this->port = $port;
+		if ($auto_connect)
+		{
+			$this->connect();
+		}
+		$this->throw_exception_on_timeout = $throw_exception_on_timeout;
+	}
+
+	public function connect()
+	{
+		if ($this->connected) return;
+		// тушим ошибки, иначе вывалится с PHP Error вместо exception
+		if (@socket_connect($this->sock, $this->host, $this->port) === FALSE) 
+		{
+			throw new Exception("Error connecting: ".socket_last_error()." ".socket_strerror(socket_last_error()));
+		}
+		$this->connected = true;
+	}
+}
+
+class UnixSocket extends Socket
+{
+	private $path;
+
+	public function __construct($path='', $throw_exception_on_timeout = false, $auto_connect = true)
+	{
+		$this->sock = socket_create(AF_UNIX, SOCK_STREAM, 0);
+		if ($this->sock === FALSE) 
+		{
+			throw new Exception("Error creating socket: ".socket_last_error()." ".socket_strerror(socket_last_error()));
+		}
+		$this->connected = false;
+		$this->path = $path;
+		if ($auto_connect)
+		{
+			$this->connect();
+		}
+		$this->throw_exception_on_timeout = $throw_exception_on_timeout;
+	}
+
+	public function connect()
+	{
+		if ($this->connected) return;
+		// тушим ошибки, иначе вывалится с PHP Error вместо exception
+		if (@socket_connect($this->sock, $this->path) === FALSE) 
+		{
+			throw new Exception("Error connecting: ".socket_last_error()." ".socket_strerror(socket_last_error()));
+		}
+		$this->connected = true;
+	}
 }

client/test_client.php

@@ -4,6 +4,9 @@ require_once('Socket.php');
 
 function makeRequest($size)
 {
+	$result = '';
+	for ($i = 0; $i < $size; $i++) $result .= chr(rand() % 256);
+	return $result;
 }
 
 function testSocket($amount, $reconnect, $size)
@@ -14,7 +17,30 @@ function testSocket($amount, $reconnect, $size)
 	{
 		if ($s === null)
 		{
-			$s = new Socket('127.0.0.1', 19876);
+			$s = new IPSocket('127.0.0.1', 19876);
+			$s->connect();
+			$current = 0;
+		}
+		$m = makeRequest($size);
+		$s->sendMessage($m);
+		$r = $s->receiveMessage();
+		$current++;
+		if ($current >= $reconnect)
+		{
+			$s = null;
+		}
+	}
+}
+
+function testUnixSocket($amount, $reconnect, $size)
+{
+	$s = null;
+	$current = 0;
+	for ($i = 0; $i < $amount; $i++)
+	{
+		if ($s === null)
+		{
+			$s = new UnixSocket('/tmp/ipc_test_socket');
 			$s->connect();
 			$current = 0;
 		}
@@ -35,9 +61,20 @@ $request_size = $argv[3];
 
 print('Testing '.$requests_amount.' requests, reconnect after '.
 		$requests_per_connect.' size '.$request_size."\n");
+
+
+print("Testing IP socket:\n");
 $start = microtime(true);
 testSocket($requests_amount, $requests_per_connect, $request_size);
 $finish = microtime(true);
 
 $elapsed = $finish - $start;
 print('Total: '.round($elapsed, 3).' seconds, '.round($requests_amount / $elapsed, 3).' RPS'."\n");
+
+print("Testing Unix socket:\n");
+$start = microtime(true);
+testUnixSocket($requests_amount, $requests_per_connect, $request_size);
+$finish = microtime(true);
+
+$elapsed = $finish - $start;
+print('Total: '.round($elapsed, 3).' seconds, '.round($requests_amount / $elapsed, 3).' RPS'."\n");

server/CMakeLists.txt

@@ -4,6 +4,8 @@ project(ipc_test_server)
 set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
 set(CMAKE_MODULE_PATH "${CMAKE_SOURCE_DIR}/../cmake")
 
+find_package(Threads REQUIRED)
+
 ## SOURCES  ####################################################################
 set(SOURCE_FILES
 
@@ -17,7 +19,7 @@ SynchronousSocket.h
 )
 ##################################################################################
 
-link_libraries(stdc++)
+link_libraries(stdc++ ${CMAKE_THREAD_LIBS_INIT})
 
 add_executable(ipc_test_server ${SOURCE_FILES} ./ipc_server.cpp)
 

server/ipc_server.cpp

@@ -6,6 +6,7 @@
 #include <sys/file.h>
 #include <errno.h>
 #include <iostream>
+#include <sys/un.h>
 
 #include "SynchronousSocket.h"
 #include "task.h"
@@ -16,6 +17,8 @@ const int LISTEN_PORT = 19876;
 const int LISTEN_BACKLOG = 10;
 const int EXIT_FAIL = 1;
 
+const char* UNIX_SOCKET_PATH = "/tmp/ipc_test_socket";
+
 void ProcessRequests(int socket_fd)
 {
 	SynchronousSocket s;
@@ -38,13 +41,13 @@ void ProcessRequests(int socket_fd)
 	s.Disconnect();
 }
 
-int main()
+void* IPSocketThread(void* arg)
 {
 	int listensock;
 	if ((listensock = socket(PF_INET, SOCK_STREAM, 0)) == -1)
 	{
 		cerr << "Could not create socket for listening...\n";
-		return EXIT_FAIL;
+		exit(EXIT_FAIL);
 	}
 	struct sockaddr_in listenaddr;
 	memset(&listenaddr, 0, sizeof(listenaddr));
@@ -54,18 +57,17 @@ int main()
 	if (bind(listensock, (struct  sockaddr *)(&listenaddr), sizeof(struct sockaddr_in)) == -1)
 	{
 		cerr << "Could not bind listening socket...\n";
-		return EXIT_FAIL;
+		exit(EXIT_FAIL);
 	}
 	if (listen(listensock, LISTEN_BACKLOG) == -1)
 	{
 		cerr << "Could not set socket into listening mode...\n";
-		return EXIT_FAIL;
+		exit(EXIT_FAIL);
 	}
 	struct sockaddr_in peeraddr;
 	socklen_t paddrlen;
 	paddrlen = sizeof(peeraddr);
 	int recsock;
-	cout << "Press Ctrl-C for exit\n";
 	cout << "Accepting connections on port " << LISTEN_PORT << "\n";           
 	while (true)
 	{
@@ -84,5 +86,100 @@ int main()
 		cerr << "Could not close listening socket...\n";
 		exit(EXIT_FAIL);
 	}
+}
+
+void* UnixSocketThread(void* arg)
+{
+	int listensock;
+	if ((listensock = socket(AF_UNIX, SOCK_STREAM, 0)) == -1)
+	{
+		cerr << "Could not create socket for listening...\n";
+		exit(EXIT_FAIL);
+	}
+	struct sockaddr_un listenaddr;
+	listenaddr.sun_family = AF_UNIX;
+	strcpy(listenaddr.sun_path, UNIX_SOCKET_PATH);
+	unlink(listenaddr.sun_path);
+	socklen_t len = strlen(listenaddr.sun_path) + sizeof(listenaddr.sun_family);
+	if (bind(listensock, (struct sockaddr*)&listenaddr, len) != 0)
+	{
+		cerr << "Error on binding socket \n";
+		perror("bind");
+		exit(EXIT_FAIL);
+	}
+	if (listen(listensock, LISTEN_BACKLOG) == -1)
+	{
+		cerr << "Could not set socket into listening mode...\n";
+		exit(EXIT_FAIL);
+	}
+	struct sockaddr_un peeraddr;
+	socklen_t paddrlen;
+	paddrlen = sizeof(peeraddr);
+	int recsock;
+	cout << "Accepting connections on path " << UNIX_SOCKET_PATH << "\n";           
+	while (true)
+	{
+		(recsock = accept(listensock, (struct sockaddr *)(&peeraddr), &paddrlen));
+		if (recsock == -1)
+		{
+			cerr << "Error accepting connection, errno = " << errno << "\n";
+			continue;
+		}
+		cout << "Accepted connection from " << peeraddr.sun_path << "\n";
+		ProcessRequests(recsock);
+	}                                                   
+	cerr << "Accept error...\n";
+	if (close(listensock) == -1)
+	{
+		cerr << "Could not close listening socket...\n";
+		exit(EXIT_FAIL);
+	}
+}
+
+void CreateSocketThread(void *(*start)(void *))
+{   
+	pthread_t thread;
+	pthread_attr_t attr;
+	if (pthread_attr_init(&attr) != 0)
+	{                    	
+		throw (string)("pthread_attr_init error");
+	}
+	if (pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED))
+	{
+		throw (string)("pthread_attr_setdetachstate error");
+	}                 
+	if (pthread_create(&thread, &attr, start, nullptr) != 0)
+	{                      
+		throw (string)("pthread_create error");
+	}                
+}
+
+void CreateUnixSocket()
+{   
+	pthread_t thread;
+	pthread_attr_t attr;
+	if (pthread_attr_init(&attr) != 0)
+	{                    	
+		throw (string)("pthread_attr_init error");
+	}
+	if (pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED))
+	{
+		throw (string)("pthread_attr_setdetachstate error");
+	}                 
+	if (pthread_create(&thread, &attr, &IPSocketThread, nullptr) != 0)
+	{                      
+		throw (string)("pthread_create error");
+	}                
+}
+
+int main()
+{
+	CreateSocketThread(&IPSocketThread);
+	CreateSocketThread(&UnixSocketThread);
+	cout << "Press Ctrl-C for exit\n";
+	while (true)
+	{
+		sleep(10);
+	}
 	return 0;
 }