redmine

Socket control API support started

bin_PROGRAMS = clsync
clsync_SOURCES = sync.c cluster.c main.c output.c fileutils.c malloc.c glibex.c
clsync_CFLAGS = $(AM_CFLAGS)
clsync_SOURCES = sync.c cluster.c main.c socket.c output.c fileutils.c malloc.c glibex.c
if SOCKET
lib_LTLIBRARIES = libclsync.la
libclsync_la_SOURCES = malloc.c libclsync.c socket.c
libclsync_la_LDFLAGS = -version-info 3:0:0
clsync_SOURCES += socket.c control.c
endif
main.o: revision.h
... ... @@ -14,14 +21,27 @@ example_DATA = \
$(wildcard $(srcdir)/examples/*.c) \
$(wildcard $(srcdir)/examples/*.sh)
REVISION=$(shell [ -d .git ] &&\
(echo -n \
'\".'$$(($$(git log 2>/dev/null \
| grep -c ^commit \
| tr -d "\n" \
)-523 \
))'\"' \
) || echo -n '"-release"' )
AM_CFLAGS := -DREVISION=$(REVISION)
clsync_includedir = $(includedir)/clsync
clsync_include_HEADERS = configuration.h clsync.h malloc.h indexes.h options.h output.h socket.h
revision.h:
(echo -n '#define REVISION "'; [ -d .git ] && \
(echo -n '.'$$(( $$(git log 2>/dev/null | grep -c ^commit | tr -d "\n") - 523 )) ) \
|| echo -n '-release'; echo '"') > $@
clsync_include_HEADERS = \
configuration.h \
clsync.h \
malloc.h \
indexes.h \
options.h \
output.h \
socket.h
doc:
doxygen .doxygen
... ... @@ -37,3 +57,4 @@ CLEANFILES = revision.h
CLEANFILES += examples/rules
clean-local:
-rm -rf examples/testdir examples/*.o examples/*.so examples/*.xz doc
... ...
... ... @@ -20,8 +20,11 @@
// control socket listen backlog (man 2 listen)
#define SOCKET_BACKLOG 2
// control socket clients limit
#define SOCKET_CLIENTS_MAX 8
// control socket connections limit in clsync
#define SOCKET_MAX_CLSYNC 8
// control socket connections limit in libclsync
#define SOCKET_MAX_LIBCLSYNC (1<<16)
// children count limit
#define MAXCHILDREN (1<<8)
... ...
... ... @@ -2,10 +2,13 @@
# Process this file with autoconf to produce a configure script.
AC_PREREQ([2.59])
AC_INIT([clsync],[0.2],[Dmitry Yu Okunev <dyokunev@ut.mephi.ru>],,[https://github.com/xaionaro/clsync])
AC_INIT([clsync],[0.3],[Dmitry Yu Okunev <dyokunev@ut.mephi.ru>],,[https://github.com/xaionaro/clsync])
AC_CONFIG_SRCDIR([sync.c])
AM_INIT_AUTOMAKE([1.11 foreign -Wall])
AM_INIT_AUTOMAKE([1.11 foreign -Wall -Wno-portability])
AC_CONFIG_HEADERS([config.h])
AM_PROG_CC_C_O
LT_INIT
dnl --enable-cluster
AC_ARG_ENABLE(cluster,
... ... @@ -34,6 +37,8 @@ AS_HELP_STRING(--enable-socket,
AS_IF([test "x$enable_socket" = "xyes"],
[CPPFLAGS+=" -DENABLE_SOCKET"])
AM_CONDITIONAL([SOCKET], [test "x$enable_socket" = "xyes"])
dnl --enable-debug
AC_ARG_ENABLE(debug,
AS_HELP_STRING(--enable-debug,
... ...
/*
clsync - file tree sync utility based on fanotify and inotify
Copyright (C) 2013 Dmitry Yu Okunev <dyokunev@ut.mephi.ru> 0x8E30679C
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "common.h"
#include <sys/un.h> // for "struct sockaddr_un"
#include "output.h"
#include "sync.h"
#include "socket.h"
#include "control.h"
static pthread_t pthread_control;
int control_procclsyncconn(socket_procconnproc_arg_t *arg, sockcmd_t *sockcmd_p) {
clsyncconn_t *clsyncconn_p = arg->clsyncconn_p;
options_t *options_p = (options_t *)arg->arg;
switch(sockcmd_p->cmd_id) {
case SOCKCMD_INFO: {
socket_send(clsyncconn_p, SOCKCMD_INFO, options_p->config_block, options_p->label, options_p->flags, options_p->flags_set);
break;
}
case SOCKCMD_DIE: {
sync_term(SIGTERM);
break;
}
}
return EINVAL;
}
static inline void closecontrol(options_t *options_p) {
if(options_p->socket) {
close(options_p->socket);
options_p->socket = 0;
}
}
int control_loop(options_t *options_p) {
pthread_t clsyncconns_threads[SOCKET_MAX_CLSYNC+1];
struct socket_procconnproc_arg clsyncconns_args[SOCKET_MAX_CLSYNC+1] = {{0}};
// Starting
printf_d("Debug2: control_loop() started (options_p->socket == %u)\n", options_p->socket);
int s;
while((s=options_p->socket)) {
// Check if the socket is still alive
if(socket_check_bysock(s)) {
printf_d("Debug: Control socket closed [case 0]: %s\n", strerror(errno));
closecontrol(options_p);
continue;
}
// Waiting for event
printf_ddd("Debug3: control_loop(): waiting for events on the socket\n");
fd_set rfds;
FD_ZERO(&rfds);
FD_SET(s, &rfds);
int count = select(s+1, &rfds, NULL, NULL, NULL);
// Processing the events
printf_dd("Debug2: control_loop(): got %i events with select()\n", count);
// Processing the events: checks
if(count == 0) {
printf_dd("Debug2: control_loop(): select() timed out.\n");
continue;
}
if(count < 0) {
printf_d("Debug: control_loop(): Got negative events count. Closing the socket.\n");
closecontrol(options_p);
continue;
}
if(!FD_ISSET(s, &rfds)) {
printf_e("Error: control_loop(): Got event, but not on the control socket. Closing the socket (cannot use \"select()\").\n");
closecontrol(options_p);
continue;
}
// Cleaning up after died connections
int i=clsyncconns_last+1;
while(i) {
i--;
switch(clsyncconns_args[i].state) {
case CLSTATE_DIED:
printf_ddd("Debug3: control_loop(): Forgeting clsyncconn #%u\n", i);
pthread_join(clsyncconns_threads[i], NULL);
clsyncconns_args[i].state = CLSTATE_NONE;
break;
default:
break;
}
}
// Processing the events: accepting new clsyncconn
clsyncconn_t *clsyncconn_p = socket_accept(s);
if(clsyncconn_p == NULL) {
if(errno == EUSERS) // Too many connections. Just ignoring the new one.
continue;
// Got unknown error. Closing control socket just in case.
printf_e("Error: control_loop(): Cannot socket_accept(): %s (errno: %i)\n", strerror(errno), errno);
closecontrol(options_p);
continue;
}
struct socket_procconnproc_arg *connproc_arg = &clsyncconns_args[clsyncconns_num];
#ifdef PARANOID
// Processing the events: checking if previous check were been made right
if(connproc_arg->state != CLSTATE_NONE) {
// This's not supposed to be
printf_e("Internal-Error: control_loop(): connproc_arg->state != CLSTATE_NONE\n");
closecontrol(options_p);
continue;
}
#endif
// Processing the events: creating a thread for new connection
printf_ddd("Debug3: control_loop(): clsyncconns_count == %u;\tclsyncconns_last == %u;\tclsyncconn_num == %u\n",
clsyncconns_count, clsyncconns_last, clsyncconns_num);
clsyncconns_last = MAX(clsyncconns_last, clsyncconns_num);
clsyncconns_count++;
connproc_arg->procfunct = control_procclsyncconn;
connproc_arg->clsyncconn_p = clsyncconn_p;
connproc_arg->arg = options_p;
connproc_arg->running = &options_p->socket;
connproc_arg->authtype = options_p->flags[SOCKETAUTH];
connproc_arg->flags = 0;
printf_dd("Debug2: control_loop(): Starting new thread for new connection.\n");
if(pthread_create(&clsyncconns_threads[clsyncconns_num], NULL, (void *(*)(void *))socket_procclsyncconn, connproc_arg)) {
printf_e("Error: control_loop(): Cannot create a thread for connection: %s (errno: %i)\n", strerror(errno), errno);
closecontrol(options_p);
continue;
}
#ifdef DEBUG
// Too prevent to often connections
sleep(1);
#endif
}
// Cleanup
printf_d("Debug2: control_loop() finished\n");
return 0;
}
int control_run(options_t *options_p) {
if(options_p->socketpath != NULL) {
int ret = 0;
// initializing clsync-socket subsystem
if((ret = socket_init()))
printf_e("Error: Cannot init clsync-sockets subsystem.\n");
// creating a simple unix socket
int s = -1;
if(!ret)
s = socket(AF_UNIX, SOCK_STREAM, 0);
// checking the path
if(!ret) {
// already exists? - unlink
if(!access(options_p->socketpath, F_OK))
if(unlink(options_p->socketpath)) {
printf_e("Error: Cannot unlink() \"%s\": %s (errno: %i).\n",
options_p->socketpath, strerror(errno), errno);
ret = errno;
}
}
// binding
if(!ret) {
struct sockaddr_un addr;
memset(&addr, 0, sizeof(addr));
addr.sun_family = AF_UNIX;
strncpy(addr.sun_path, options_p->socketpath, sizeof(addr.sun_path)-1);
if(bind(s, (struct sockaddr *)&addr, sizeof(addr))) {
printf_e("Error: Cannot bind() on address \"%s\": %s (errno: %i).\n",
options_p->socketpath, strerror(errno), errno);
ret = errno;
}
}
// starting to listening
if(!ret) {
if(listen(s, SOCKET_BACKLOG)) {
printf_e("Error: Cannot listen() on address \"%s\": %s (errno: %i).\n",
options_p->socketpath, strerror(errno), errno);
ret = errno;
}
}
// fixing privileges
if(!ret) {
if(options_p->flags[SOCKETMOD])
if(chmod(options_p->socketpath, options_p->socketmod)) {
printf_e("Error, Cannot chmod(\"%s\", %o): %s (errno: %i)\n",
options_p->socketpath, options_p->socketmod, strerror(errno), errno);
ret = errno;
}
if(options_p->flags[SOCKETOWN])
if(chown(options_p->socketpath, options_p->socketuid, options_p->socketgid)) {
printf_e("Error, Cannot chown(\"%s\", %u, %u): %s (errno: %i)\n",
options_p->socketpath, options_p->socketuid, options_p->socketgid, strerror(errno), errno);
ret = errno;
}
}
// finish
if(ret) {
close(s);
return ret;
}
options_p->socket = s;
printf_dd("Debug2: control_run(): options_p->socket = %u\n", options_p->socket);
ret = pthread_create(&pthread_control, NULL, (void *(*)(void *))control_loop, options_p);
}
return 0;
}
int control_cleanup(options_t *options_p) {
if(options_p->socketpath != NULL) {
unlink(options_p->socketpath);
closecontrol(options_p);
// TODO: kill pthread_control and join
// pthread_join(pthread_control, NULL);
socket_deinit();
}
return 0;
}
... ...
/*
clsync - file tree sync utility based on fanotify and inotify
Copyright (C) 2013 Dmitry Yu Okunev <dyokunev@ut.mephi.ru> 0x8E30679C
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
extern int control_run(options_t *options_p);
extern int control_cleanup(options_t *options_p);
... ...
clsync (0.3-1) unstable; urgency=low
* Added support of control socket
-- Dmitry Yu Okunev <dyokunev@ut.mephi.ru> Thu, 9 Jan 2014 15:38:33 +0400
clsync (0.2.1-1) unstable; urgency=low
* New upstream version
... ...
... ... @@ -20,3 +20,43 @@ Description: live sync tool based on inotify, written in GNU C
This utility is much more lightweight than competitors and supports such
features as separate queue for big files, regex file filter,
multi-threading.
Package: clsync-doc
Architecture: any
Depends: ${shlibs:Depends}, ${misc:Depends}
Description: examples and other documentation for clsync
Clsync recursively watches for source directory and executes external
program to sync the changes. Clsync is adapted to use together with rsync.
This utility is much more lightweight than competitors and supports such
features as separate queue for big files, regex file filter,
multi-threading.
Package: clsync-dev
Architecture: any
Depends: ${shlibs:Depends}, ${misc:Depends}
Description: development files for clsync
Clsync recursively watches for source directory and executes external
program to sync the changes. Clsync is adapted to use together with rsync.
This utility is much more lightweight than competitors and supports such
features as separate queue for big files, regex file filter,
multi-threading.
Package: libclsync
Architecture: any
Depends: ${shlibs:Depends}, ${misc:Depends}
Description: clsync control socket library
Clsync recursively watches for source directory and executes external
program to sync the changes. Clsync is adapted to use together with rsync.
This utility is much more lightweight than competitors and supports such
features as separate queue for big files, regex file filter,
multi-threading.
Package: libclsync-dev
Architecture: any
Depends: ${shlibs:Depends}, ${misc:Depends}
Description: development files for libclsync
Clsync recursively watches for source directory and executes external
program to sync the changes. Clsync is adapted to use together with rsync.
This utility is much more lightweight than competitors and supports such
features as separate queue for big files, regex file filter,
multi-threading.
... ...
... ... @@ -12,6 +12,9 @@
%:
dh $@ --parallel --with autoreconf
override_dh_auto_install:
dh_auto_install
-rm --verbose debian/clsync/usr/share/doc/clsync/LICENSE
override_dh_auto_configure:
dh_auto_configure -- --enable-socket=yes
#override_dh_auto_install:
# dh_auto_install
# -rm --verbose debian/clsync/usr/share/doc/clsync/LICENSE
... ...
... ... @@ -28,8 +28,9 @@
#include "malloc.h"
#include "cluster.h"
#include "fileutils.h"
#include "socket.h"
#include "revision.h"
//#include "revision.h"
static const struct option long_options[] =
{
... ...
... ... @@ -89,13 +89,6 @@ enum flags_enum {
};
typedef enum flags_enum flags_t;
enum sockauth_id {
SOCKAUTH_UNSET = 0,
SOCKAUTH_NULL,
SOCKAUTH_PAM,
};
typedef enum sockauth_id sockauth_id_t;
enum mode_id {
MODE_UNSET = 0,
MODE_SIMPLE,
... ...
This diff is collapsed. Click to expand it.
... ... @@ -17,18 +17,19 @@
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define SOCKET_DEFAULT_PROT 0
#define SOCKET_DEFAULT_SUBPROT SUBPROT0_TEXT
#define SOCKET_DEFAULT_PROT 0
#define SOCKET_DEFAULT_SUBPROT SUBPROT0_TEXT
// buffer size
#define SOCKET_BUFSIZ (1<<12)
struct client {
struct clsyncconn {
int sock;
int num;
uint16_t prot;
uint16_t subprot;
};
typedef struct client client_t;
typedef struct clsyncconn clsyncconn_t;
enum subprot0 {
SUBPROT0_TEXT,
... ... @@ -41,14 +42,14 @@ struct sockcmd_negotiation {
uint16_t subprot;
};
enum client_state {
enum clsyncconn_state {
CLSTATE_NONE = 0,
CLSTATE_AUTH,
CLSTATE_MAIN,
CLSTATE_DYING,
CLSTATE_DIED,
};
typedef enum client_state client_state_t;
typedef enum clsyncconn_state clsyncconn_state_t;
enum sockcmd_id {
SOCKCMD_NEGOTIATION = 00,
... ... @@ -66,30 +67,6 @@ enum sockcmd_id {
};
typedef enum sockcmd_id sockcmd_id_t;
static char *const textmessage_args[] = {
[SOCKCMD_NEGOTIATION] = "%u",
[SOCKCMD_ACK] = "%03u %lu",
[SOCKCMD_EINVAL] = "%03u %lu",
[SOCKCMD_VERSION] = "%u %u",
[SOCKCMD_INFO] = "%s\003/ %s\003/ %x %x",
[SOCKCMD_UNKNOWNCMD] = "%03u %lu",
[SOCKCMD_INVALIDCMDID] = "%lu",
};
static char *const textmessage_descr[] = {
[SOCKCMD_NEGOTIATION] = "Protocol version is %u.",
[SOCKCMD_ACK] = "Acknowledged command: id == %03u; num == %lu.",
[SOCKCMD_EINVAL] = "Rejected command: id == %03u; num == %lu. Invalid arguments: %s.",
[SOCKCMD_LOGIN] = "Enter your login and password, please.",
[SOCKCMD_UNEXPECTEDEND] = "Need to go, sorry. :)",
[SOCKCMD_DIE] = "Okay :(",
[SOCKCMD_BYE] = "Bye.",
[SOCKCMD_VERSION] = "clsync v%u.%u.",
[SOCKCMD_INFO] = "config_block == \"%s\"; label == \"%s\"; flags == %x; flags_set == %x.",
[SOCKCMD_UNKNOWNCMD] = "Unknown command.",
[SOCKCMD_INVALIDCMDID] = "Invalid command id. Required: 0 <= cmd_id < 1000.",
};
struct sockcmd {
uint64_t cmd_num;
uint16_t cmd_id;
... ... @@ -98,7 +75,45 @@ struct sockcmd {
};
typedef struct sockcmd sockcmd_t;
extern int socket_run(options_t *options_p);
extern int socket_cleanup(options_t *options_p);
enum sockprocflags {
SOCKPROCFLAG_NONE = 0,
SOCKPROCFLAG_OVERRIDECOMMON,
};
typedef enum sockprocflags sockprocflags_t;
enum sockauth_id {
SOCKAUTH_UNSET = 0,
SOCKAUTH_NULL,
SOCKAUTH_PAM,
};
typedef enum sockauth_id sockauth_id_t;
struct socket_procconnproc_arg;
typedef int (*clsyncconn_procfunct_t)(struct socket_procconnproc_arg *, sockcmd_t *);
struct socket_procconnproc_arg {
clsyncconn_procfunct_t procfunct;
clsyncconn_t *clsyncconn_p;
void *arg;
clsyncconn_state_t state;
sockauth_id_t authtype;
int *running; // Pointer to interger with non-zero value to continue running
sockprocflags_t flags;
};
typedef struct socket_procconnproc_arg socket_procconnproc_arg_t;
extern int socket_send(clsyncconn_t *clsyncconn, sockcmd_id_t cmd_id, ...);
extern int socket_recv(clsyncconn_t *clsyncconn, sockcmd_t *sockcmd);
extern int socket_check_bysock(int sock);
extern int socket_close(clsyncconn_t *clsyncconn);
extern clsyncconn_t *socket_accept(int sock);
extern int socket_init();
extern int socket_deinit();
extern int socket_procclsyncconn(socket_procconnproc_arg_t *arg);
extern int clsyncconns_num;
extern int clsyncconns_count;
extern int clsyncconns_last;
extern const char *const textmessage_args[];
extern const char *const textmessage_descr[];
... ...
... ... @@ -26,7 +26,7 @@
#include "cluster.h"
#include "sync.h"
#include "glibex.h"
#include "socket.h"
#include "control.h"
#include <dlfcn.h>
... ... @@ -3190,7 +3190,7 @@ int sync_run(options_t *options_p) {
#ifdef ENABLE_SOCKET
// Creating control socket
if(options_p->socketpath != NULL)
ret = socket_run(options_p);
ret = control_run(options_p);
#endif
if(!options_p->flags[ONLYINITSYNC]) {
... ... @@ -3214,7 +3214,7 @@ int sync_run(options_t *options_p) {
#ifdef ENABLE_SOCKET
// Removing control socket
if(options_p->socketpath != NULL)
socket_cleanup(options_p);
control_cleanup(options_p);
#endif
printf_d("Debug2: sync_run(): killing sighandler\n");
... ...