/*

clsync - file tree sync utility based on inotify/kqueue

Copyright (C) 2013-2014 Dmitry Yu Okunev <[email protected]> 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"

#if KQUEUE_SUPPORT

# include "mon_kqueue.h"

#endif

#if INOTIFY_SUPPORT

# include "mon_inotify.h"

#endif

#if FANOTIFY_SUPPORT

# include "mon_fanotify.h"

#endif

#if BSM_SUPPORT

# include "mon_bsm.h"

# include <bsm/audit_kevents.h>

#endif

#if GIO_SUPPORT

# include <gio/gio.h>

# include "mon_gio.h"

#endif

#include "main.h"

#include "error.h"

#include "fileutils.h"

#include "malloc.h"

#include "cluster.h"

#include "sync.h"

#include "glibex.h"

#include "control.h"

#include "indexes.h"

#include "privileged.h"

#include "rules.h"

#if CGROUP_SUPPORT

# include "cgroup.h"

#endif

#include <stdio.h>

#include <dlfcn.h>

pthread_t pthread_sighandler;

// seqid - is a counter of main loop. But it may overflow and it's required to compare

// seqid-values anyway.

// So if (a-b) is too big, let's assume, that "b<a".

#define SEQID_WINDOW (((unsigned int)~0)>>1)

#define SEQID_EQ(a, b) ((a)==(b))

#define SEQID_GE(a, b) ((a)-(b) < SEQID_WINDOW)

#define SEQID_LE(a, b) ((b)-(a) < SEQID_WINDOW)

#define SEQID_GT(a, b) ((!SEQID_EQ(a, b)) && (SEQID_GE(a, b)))

#define SEQID_LT(a, b) ((!SEQID_EQ(a, b)) && (SEQID_LE(a, b)))

static unsigned int _sync_seqid_value = 0;

static inline unsigned int sync_seqid()

{

return _sync_seqid_value++;

}

static inline void setenv_iteration ( uint32_t iteration_num )

{

char iterations[sizeof ( "4294967296" )]; // 4294967296 == 2**32

sprintf ( iterations, "%i", iteration_num );

setenv ( "CLSYNC_ITERATION", iterations, 1 );

return;

}

static inline void finish_iteration ( ctx_t *ctx_p )

{

if ( ctx_p->iteration_num < ( typeof ( ctx_p->iteration_num ) ) ~0 ) // ~0 is the max value for unsigned variables

ctx_p->iteration_num++;

#ifdef THREADING_SUPPORT

if ( !ctx_p->flags[THREADING] )

#endif

setenv_iteration ( ctx_p->iteration_num );

debug ( 3, "next iteration: %u/%u",

ctx_p->iteration_num, ctx_p->flags[MAXITERATIONS] );

return;

}

gpointer eidup ( gpointer ei_gp )

{

eventinfo_t *ei = ( eventinfo_t * ) ei_gp;

eventinfo_t *ei_dup = ( eventinfo_t * ) xmalloc ( sizeof ( *ei ) );

memcpy ( ei_dup, ei, sizeof ( *ei ) );

return ( gpointer ) ei_dup;

}

static inline void evinfo_merge ( ctx_t *ctx_p, eventinfo_t *evinfo_dst, eventinfo_t *evinfo_src )

{

debug ( 3, "evinfo_dst: seqid_min == %u; seqid_max == %u; objtype_old == %i; objtype_new == %i; \t"

"evinfo_src: seqid_min == %u; seqid_max == %u; objtype_old == %i; objtype_new == %i",

evinfo_dst->seqid_min, evinfo_dst->seqid_max, evinfo_dst->objtype_old, evinfo_dst->objtype_new,

evinfo_src->seqid_min, evinfo_src->seqid_max, evinfo_src->objtype_old, evinfo_src->objtype_new

);

#if KQUEUE_SUPPORT | INOTIFY_SUPPORT

switch ( ctx_p->flags[MONITOR] ) {

#ifdef KQUEUE_SUPPORT

case NE_KQUEUE:

#endif

#ifdef INOTIFY_SUPPORT

case NE_INOTIFY:

#endif

evinfo_dst->evmask |= evinfo_src->evmask;

break;

}

#endif

evinfo_dst->flags |= evinfo_src->flags;

if ( SEQID_LE ( evinfo_src->seqid_min, evinfo_dst->seqid_min ) ) {

evinfo_dst->objtype_old = evinfo_src->objtype_old;

evinfo_dst->seqid_min = evinfo_src->seqid_min;

}

if ( SEQID_GE ( evinfo_src->seqid_max, evinfo_dst->seqid_max ) ) {

evinfo_dst->objtype_new = evinfo_src->objtype_new;

evinfo_dst->seqid_max = evinfo_src->seqid_max;

switch ( ctx_p->flags[MONITOR] ) {

#ifdef GIO_SUPPORT

case NE_GIO:

evinfo_dst->evmask = evinfo_src->evmask;

break;

#endif

#ifdef BSM_SUPPORT

case NE_BSM:

case NE_BSM_PREFETCH:

evinfo_dst->evmask = evinfo_src->evmask;

break;

#endif

default:

break;

}

}

return;

}

static inline int _exitcode_process ( ctx_t *ctx_p, int exitcode )

{

if ( ctx_p->isignoredexitcode[ ( unsigned char ) exitcode] )

return 0;

if ( exitcode && ! ( ( ctx_p->flags[MODE] == MODE_RSYNCDIRECT ) && ( exitcode == 24 ) ) ) {

error ( "Got non-zero exitcode %i from __sync_exec().", exitcode );

return exitcode;

}

return 0;

}

int exitcode_process ( ctx_t *ctx_p, int exitcode )

{

int err = _exitcode_process ( ctx_p, exitcode );

if ( err ) error ( "Got error-report from exitcode_process().\nExitcode is %i, strerror(%i) returns \"%s\". However strerror() is not ensures compliance "

"between exitcode and error description for every utility. So, e.g if you're using rsync, you should look for the error description "

"into rsync's manpage (\"man 1 rsync\"). Also some advices about diagnostics can be found in clsync's manpage (\"man 1 clsync\", see DIAGNOSTICS)",

exitcode, exitcode, strerror ( exitcode ) );

return err;

}

threadsinfo_t *thread_info() // TODO: optimize this

{

static threadsinfo_t threadsinfo = {0};

if ( !threadsinfo.mutex_init ) {

int i = 0;

while ( i < PTHREAD_MUTEX_MAX ) {

if ( pthread_mutex_init ( &threadsinfo.mutex[i], NULL ) ) {

error ( "Cannot pthread_mutex_init()." );

return NULL;

}

if ( pthread_cond_init ( &threadsinfo.cond [i], NULL ) ) {

error ( "Cannot pthread_cond_init()." );

return NULL;

}

i++;

}

threadsinfo.mutex_init++;

}

return &threadsinfo;

}

#ifdef THREADING_SUPPORT

#define thread_info_lock() _thread_info_lock(__func__)

static inline threadsinfo_t *_thread_info_lock ( const char *const function_name )

{

threadsinfo_t *threadsinfo_p = thread_info();

debug ( 4, "used by %s()", function_name );

pthread_mutex_lock ( &threadsinfo_p->mutex[PTHREAD_MUTEX_THREADSINFO] );

return threadsinfo_p;

}

#define thread_info_unlock(...) _thread_info_unlock(__func__, __VA_ARGS__)

static inline int _thread_info_unlock ( const char *const function_name, int rc )

{

threadsinfo_t *threadsinfo_p = thread_info();

debug ( 4, "used by %s()", function_name );

pthread_mutex_unlock ( &threadsinfo_p->mutex[PTHREAD_MUTEX_THREADSINFO] );

return rc;

}

int threads_foreach ( int ( *funct ) ( threadinfo_t *, void * ), state_t state, void *arg )

{

int i, rc;

threadsinfo_t *threadsinfo_p = thread_info_lock();

#ifdef PARANOID

if ( threadsinfo_p == NULL )

return thread_info_unlock ( EINVAL );

#endif

rc = 0;

i = 0;

while ( i < threadsinfo_p->used ) {

threadinfo_t *threadinfo_p = &threadsinfo_p->threads[i++];

if ( ( state == STATE_UNKNOWN ) || ( threadinfo_p->state == state ) ) {

if ( ( rc = funct ( threadinfo_p, arg ) ) )

break;

}

}

return thread_info_unlock ( rc );

}

time_t thread_nextexpiretime()

{

time_t nextexpiretime = 0;

threadsinfo_t *threadsinfo_p = thread_info_lock();

#ifdef PARANOID

if ( threadsinfo_p == NULL )

return thread_info_unlock ( 0 );

#endif

int thread_num = threadsinfo_p->used;

while ( thread_num-- ) {

threadinfo_t *threadinfo_p = &threadsinfo_p->threads[thread_num];

debug ( 3, "threadsinfo_p->threads[%i].state == %i;\tthreadsinfo_p->threads[%i].pthread == %p;\tthreadsinfo_p->threads[%i].expiretime == %i",

thread_num, threadinfo_p->state, thread_num, threadinfo_p->pthread, thread_num, threadinfo_p->expiretime );

if ( threadinfo_p->state == STATE_EXIT )

continue;

if ( threadinfo_p->expiretime ) {

if ( nextexpiretime )

nextexpiretime = MIN ( nextexpiretime, threadinfo_p->expiretime );

else

nextexpiretime = threadinfo_p->expiretime;

}

}

thread_info_unlock ( 0 );

debug ( 3, "nextexpiretime == %i", nextexpiretime );

return nextexpiretime;

}

threadinfo_t *thread_new()

{

threadsinfo_t *threadsinfo_p = thread_info_lock();

#ifdef PARANOID

if ( threadsinfo_p == NULL ) {

thread_info_unlock ( 0 );

return NULL;

}

#endif

int thread_num;

threadinfo_t *threadinfo_p;

if ( threadsinfo_p->stacklen ) {

threadinfo_p = threadsinfo_p->threadsstack[--threadsinfo_p->stacklen];

thread_num = threadinfo_p->thread_num;

} else {

if ( threadsinfo_p->used >= threadsinfo_p->allocated ) {

threadsinfo_p->allocated += ALLOC_PORTION;

debug ( 2, "Reallocated memory for threadsinfo -> %i.", threadsinfo_p->allocated );

threadsinfo_p->threads = ( threadinfo_t * ) xrealloc ( ( char * ) threadsinfo_p->threads,

sizeof ( *threadsinfo_p->threads ) * ( threadsinfo_p->allocated + 2 ) );

threadsinfo_p->threadsstack = ( threadinfo_t ** ) xrealloc ( ( char * ) threadsinfo_p->threadsstack,

sizeof ( *threadsinfo_p->threadsstack ) * ( threadsinfo_p->allocated + 2 ) );

}

thread_num = threadsinfo_p->used++;

threadinfo_p = &threadsinfo_p->threads[thread_num];

}

#ifdef PARANOID

memset ( threadinfo_p, 0, sizeof ( *threadinfo_p ) );

#else

threadinfo_p->expiretime = 0;

threadinfo_p->errcode = 0;

threadinfo_p->exitcode = 0;

#endif

threadinfo_p->thread_num = thread_num;

threadinfo_p->state = STATE_RUNNING;

debug ( 2, "thread_new -> thread_num: %i; used: %i", thread_num, threadsinfo_p->used );

thread_info_unlock ( 0 );

return threadinfo_p;

}

int thread_del_bynum ( int thread_num )

{

debug ( 2, "thread_del_bynum(%i)", thread_num );

threadsinfo_t *threadsinfo_p = thread_info_lock();

#ifdef PARANOID

if ( threadsinfo_p == NULL )

return thread_info_unlock ( errno );

#endif

if ( thread_num >= threadsinfo_p->used )

return thread_info_unlock ( EINVAL );

threadinfo_t *threadinfo_p = &threadsinfo_p->threads[thread_num];

threadinfo_p->state = STATE_EXIT;

char **ptr = threadinfo_p->argv;

if ( ptr != NULL ) {

while ( *ptr )

free ( * ( ptr++ ) );

free ( threadinfo_p->argv );

}

if ( thread_num == ( threadsinfo_p->used - 1 ) ) {

threadsinfo_p->used--;

debug ( 3, "thread_del_bynum(%i): there're %i threads left (#0).", thread_num, threadsinfo_p->used - threadsinfo_p->stacklen );

return thread_info_unlock ( 0 );

}

threadinfo_t *t = &threadsinfo_p->threads[threadsinfo_p->used - 1];

if ( t->state == STATE_EXIT ) {

threadsinfo_p->used--;

debug ( 3, "%i [%p] -> %i [%p]; left: %i",

threadsinfo_p->used, t->pthread, thread_num, threadinfo_p->pthread, threadsinfo_p->used - threadsinfo_p->stacklen );

memcpy ( threadinfo_p, t, sizeof ( *threadinfo_p ) );

} else {

#ifdef PARANOID

if ( threadsinfo_p->stacklen >= threadsinfo_p->allocated ) {

error ( "Threads metadata structures pointers stack overflowed!" );

return thread_info_unlock ( EINVAL );

}

#endif

threadsinfo_p->threadsstack[threadsinfo_p->stacklen++] = threadinfo_p;

}

debug ( 3, "thread_del_bynum(%i): there're %i threads left (#1).", thread_num, threadsinfo_p->used - threadsinfo_p->stacklen );

return thread_info_unlock ( 0 );

}

int thread_gc ( ctx_t *ctx_p )

{

int thread_num;

time_t tm = time ( NULL );

debug ( 3, "tm == %i; thread %p", tm, pthread_self() );

if ( !ctx_p->flags[THREADING] )

return 0;

threadsinfo_t *threadsinfo_p = thread_info_lock();

#ifdef PARANOID

if ( threadsinfo_p == NULL )

return thread_info_unlock ( errno );

#endif

debug ( 2, "There're %i threads.", threadsinfo_p->used );

thread_num = -1;

while ( ++thread_num < threadsinfo_p->used ) {

int err;

threadinfo_t *threadinfo_p = &threadsinfo_p->threads[thread_num];

debug ( 3, "Trying thread #%i (==%i) (state: %i; expire at: %i, now: %i, exitcode: %i, errcode: %i; i_p: %p; p: %p).",

thread_num, threadinfo_p->thread_num, threadinfo_p->state, threadinfo_p->expiretime, tm, threadinfo_p->exitcode,

threadinfo_p->errcode, threadinfo_p, threadinfo_p->pthread );

if ( threadinfo_p->state == STATE_EXIT )

continue;

if ( threadinfo_p->expiretime && ( threadinfo_p->expiretime <= tm ) ) {

if ( pthread_tryjoin_np ( threadinfo_p->pthread, NULL ) ) { // TODO: check this pthread_tryjoin_np() on error returnings

error ( "Debug3: thread_gc(): Thread #%i is alive too long: %lu <= %lu (started at %lu)", thread_num, threadinfo_p->expiretime, tm, threadinfo_p->starttime );

return thread_info_unlock ( ETIME );

}

}

#ifndef VERYPARANOID

if ( threadinfo_p->state != STATE_TERM ) {

debug ( 3, "Thread #%i is busy, skipping (#0).", thread_num );

continue;

}

#endif

debug ( 3, "Trying to join thread #%i: %p", thread_num, threadinfo_p->pthread );

#ifndef VERYPARANOID

switch ( ( err = pthread_join ( threadinfo_p->pthread, NULL ) ) ) {

#else

switch ( ( err = pthread_tryjoin_np ( threadinfo_p->pthread, NULL ) ) ) {

case EBUSY:

debug ( 3, "Thread #%i is busy, skipping (#1).", thread_num );

continue;

#endif

case EDEADLK:

case EINVAL:

case 0:

debug ( 3, "Thread #%i is finished with exitcode %i (errcode %i), deleting. threadinfo_p == %p",

thread_num, threadinfo_p->exitcode, threadinfo_p->errcode, threadinfo_p );

break;

default:

error ( "Got error while pthread_join() or pthread_tryjoin_np().", strerror ( err ), err );

return thread_info_unlock ( errno );

}

if ( threadinfo_p->errcode ) {

error ( "Got error from thread #%i: errcode %i.", thread_num, threadinfo_p->errcode );

thread_info_unlock ( 0 );

thread_del_bynum ( thread_num );

return threadinfo_p->errcode;

}

thread_info_unlock ( 0 );

if ( thread_del_bynum ( thread_num ) )

return errno;

thread_info_lock();

}

debug ( 3, "There're %i threads left.", threadsinfo_p->used - threadsinfo_p->stacklen );

return thread_info_unlock ( 0 );

}

int thread_cleanup ( ctx_t *ctx_p )

{

( void ) ctx_p;

debug ( 3, "" );

threadsinfo_t *threadsinfo_p = thread_info_lock();

#ifdef PARANOID

if ( threadsinfo_p == NULL )

return thread_info_unlock ( errno );

#endif

// Waiting for threads:

debug ( 1, "There're %i opened threads. Waiting.", threadsinfo_p->used );

while ( threadsinfo_p->used ) {

// int err;

threadinfo_t *threadinfo_p = &threadsinfo_p->threads[--threadsinfo_p->used];

if ( threadinfo_p->state == STATE_EXIT )

continue;

//pthread_kill(threadinfo_p->pthread, SIGTERM);

debug ( 1, "killing pid %i with SIGTERM", threadinfo_p->child_pid );

kill ( threadinfo_p->child_pid, SIGTERM );

pthread_join ( threadinfo_p->pthread, NULL );

debug ( 2, "thread #%i exitcode: %i", threadsinfo_p->used, threadinfo_p->exitcode );

/*

if(threadinfo_p->callback)

if((err=threadinfo_p->callback(ctx_p, threadinfo_p->argv)))

warning("Got error from callback function.", strerror(err), err);

*/

char **ptr = threadinfo_p->argv;

while ( *ptr )

free ( * ( ptr++ ) );

free ( threadinfo_p->argv );

}

debug ( 3, "All threads are closed." );

// Freeing

if ( threadsinfo_p->allocated ) {

free ( threadsinfo_p->threads );

free ( threadsinfo_p->threadsstack );

}

if ( threadsinfo_p->mutex_init ) {

int i = 0;

while ( i < PTHREAD_MUTEX_MAX ) {

pthread_mutex_destroy ( &threadsinfo_p->mutex[i] );

pthread_cond_destroy ( &threadsinfo_p->cond [i] );

i++;

}

}

#ifdef PARANOID

// Reseting

memset ( threadsinfo_p, 0, sizeof ( *threadsinfo_p ) ); // Just in case;

#endif

debug ( 3, "done." );

return thread_info_unlock ( 0 );

}

#endif

volatile state_t *state_p = NULL;

volatile int exitcode = 0;

#define SHOULD_THREAD(ctx_p) ((ctx_p->flags[THREADING] != PM_OFF) && (ctx_p->flags[THREADING] != PM_SAFE || ctx_p->iteration_num))

int exec_argv ( char **argv, int *child_pid )

{

debug ( 3, "Thread %p.", pthread_self() );

pid_t pid;

int status;

// Forking

pid = privileged_fork_execvp ( argv[0], ( char *const * ) argv );

// debug(3, "After fork thread %p"")".", pthread_self() );

debug ( 3, "Child pid is %u", pid );

// Setting *child_pid value

if ( child_pid )

*child_pid = pid;

// Waiting for process end

#ifdef VERYPARANOID

sigset_t sigset_exec, sigset_old;

sigemptyset ( &sigset_exec );

sigaddset ( &sigset_exec, SIGUSR_BLOPINT );

pthread_sigmask ( SIG_BLOCK, &sigset_exec, &sigset_old );

#endif

// debug(3, "Pre-wait thread %p"")".", pthread_self() );

if ( privileged_waitpid ( pid, &status, 0 ) != pid ) {

switch ( errno ) {

case ECHILD:

debug ( 2, "Child %u is already dead.", pid );

break;

default:

error ( "Cannot waitid()." );

return errno;

}

}

// debug(3, "After-wait thread %p"")".", pthread_self() );

#ifdef VERYPARANOID

pthread_sigmask ( SIG_SETMASK, &sigset_old, NULL );

#endif

// Return

int exitcode = WEXITSTATUS ( status );

debug ( 3, "execution completed with exitcode %i", exitcode );

return exitcode;

}

#ifdef THREADING_SUPPORT

static inline int thread_exit ( threadinfo_t *threadinfo_p, int exitcode )

{

int err = 0;

threadinfo_p->exitcode = exitcode;

#if _DEBUG_FORCE | VERYPARANOID

if ( threadinfo_p->pthread != pthread_self() ) {

error ( "pthread id mismatch! (i_p->p) %p != (p) %p""", threadinfo_p->pthread, pthread_self() );

return EINVAL;

}

#endif

if ( threadinfo_p->callback ) {

if ( threadinfo_p->ctx_p->flags[DEBUG] > 2 ) {

debug ( 3, "thread %p, argv: ", threadinfo_p->pthread );

char **argv = threadinfo_p->argv;

while ( *argv ) {

debug ( 3, "\t%p == %s", *argv, *argv );

argv++;

}

}

if ( ( err = threadinfo_p->callback ( threadinfo_p->ctx_p, threadinfo_p->callback_arg ) ) ) {

error ( "Got error from callback function.", strerror ( err ), err );

threadinfo_p->errcode = err;

}

}

// Notifying the parent-thread, that it's time to collect garbage threads

threadinfo_p->state = STATE_TERM;

debug ( 3, "thread %p is sending signal to sighandler to call GC", threadinfo_p->pthread );

return pthread_kill ( pthread_sighandler, SIGUSR_THREAD_GC );

}

#endif

static inline void so_call_sync_finished ( int n, api_eventinfo_t *ei )

{

int i = 0;

api_eventinfo_t *ei_i = ei;

while ( i < n ) {

#ifdef PARANOID

if ( ei_i->path == NULL ) {

warning ( "ei_i->path == NULL" );

i++;

continue;

}

#endif

free ( ( char * ) ei_i->path );

ei_i++;

i++;

}

if ( ei != NULL )

free ( ei );

return;

}

#ifdef THREADING_SUPPORT

int so_call_sync_thread ( threadinfo_t *threadinfo_p )

{

debug ( 3, "thread_num == %i; threadinfo_p == %p; i_p->pthread %p; thread %p",

threadinfo_p->thread_num, threadinfo_p, threadinfo_p->pthread, pthread_self() );

ctx_t *ctx_p = threadinfo_p->ctx_p;

int n = threadinfo_p->n;

api_eventinfo_t *ei = threadinfo_p->ei;

int err = 0, rc = 0, try_again = 0;

do {

try_again = 0;

threadinfo_p->try_n++;

rc = ctx_p->handler_funct.sync ( n, ei );

if ( ( err = exitcode_process ( threadinfo_p->ctx_p, rc ) ) ) {

try_again = ( ( !ctx_p->retries ) || ( threadinfo_p->try_n < ctx_p->retries ) ) && ( ctx_p->state != STATE_TERM ) && ( ctx_p->state != STATE_EXIT );

warning ( "Bad exitcode %i (errcode %i). %s.", rc, err, try_again ? "Retrying" : "Give up" );

if ( try_again ) {

debug ( 2, "Sleeping for %u seconds before the retry.", ctx_p->syncdelay );

sleep ( ctx_p->syncdelay );

}

}

} while ( err && ( ( !ctx_p->retries ) || ( threadinfo_p->try_n < ctx_p->retries ) ) && ( ctx_p->state != STATE_TERM ) && ( ctx_p->state != STATE_EXIT ) );

if ( err && !ctx_p->flags[IGNOREFAILURES] ) {

error ( "Bad exitcode %i (errcode %i)", rc, err );

threadinfo_p->errcode = err;

}

so_call_sync_finished ( n, ei );

if ( ( err = thread_exit ( threadinfo_p, rc ) ) ) {

exitcode = err; // This's global variable "exitcode"

pthread_kill ( pthread_sighandler, SIGTERM );

}

return rc;

}

#endif

static inline int so_call_sync ( ctx_t *ctx_p, indexes_t *indexes_p, int n, api_eventinfo_t *ei )

{

debug ( 2, "n == %i", n );

#ifdef THREADING_SUPPORT

if ( !SHOULD_THREAD ( ctx_p ) ) {

#endif

int rc = 0, ret = 0, err = 0;

int try_n = 0, try_again;

state_t status = STATE_UNKNOWN;

// indexes_p->nonthreaded_syncing_fpath2ei_ht = g_hash_table_dup(indexes_p->fpath2ei_ht, g_str_hash, g_str_equal, free, free, (gpointer(*)(gpointer))strdup, eidup);

indexes_p->nonthreaded_syncing_fpath2ei_ht = indexes_p->fpath2ei_ht;

do {

try_again = 0;

try_n++;

alarm ( ctx_p->synctimeout );

rc = ctx_p->handler_funct.sync ( n, ei );

alarm ( 0 );

if ( ( err = exitcode_process ( ctx_p, rc ) ) ) {

if ( ( try_n == 1 ) && ( ctx_p->state != STATE_TERM ) && ( ctx_p->state != STATE_EXIT ) ) {

status = ctx_p->state;

ctx_p->state = STATE_SYNCHANDLER_ERR;

main_status_update ( ctx_p );

}

try_again = ( ( !ctx_p->retries ) || ( try_n < ctx_p->retries ) ) && ( ctx_p->state != STATE_TERM ) && ( ctx_p->state != STATE_EXIT );

warning ( "Bad exitcode %i (errcode %i). %s.", rc, err, try_again ? "Retrying" : "Give up" );

if ( try_again ) {

debug ( 2, "Sleeping for %u seconds before the retry.", ctx_p->syncdelay );

sleep ( ctx_p->syncdelay );

}

}

} while ( err && ( ( !ctx_p->retries ) || ( try_n < ctx_p->retries ) ) && ( ctx_p->state != STATE_TERM ) && ( ctx_p->state != STATE_EXIT ) );

if ( err && !ctx_p->flags[IGNOREFAILURES] ) {

error ( "Bad exitcode %i (errcode %i)", rc, err );

ret = err;

} else if ( status != STATE_UNKNOWN ) {

ctx_p->state = status;

main_status_update ( ctx_p );

}

// g_hash_table_destroy(indexes_p->nonthreaded_syncing_fpath2ei_ht);

indexes_p->nonthreaded_syncing_fpath2ei_ht = NULL;

so_call_sync_finished ( n, ei );

return ret;

#ifdef THREADING_SUPPORT

}

threadinfo_t *threadinfo_p = thread_new();

if ( threadinfo_p == NULL )

return errno;

threadinfo_p->try_n = 0;

threadinfo_p->callback = NULL;

threadinfo_p->argv = NULL;

threadinfo_p->ctx_p = ctx_p;

threadinfo_p->starttime = time ( NULL );

threadinfo_p->fpath2ei_ht = g_hash_table_dup ( indexes_p->fpath2ei_ht, g_str_hash, g_str_equal, free, free, ( gpointer ( * ) ( gpointer ) ) strdup, eidup );

threadinfo_p->n = n;

threadinfo_p->ei = ei;

threadinfo_p->iteration = ctx_p->iteration_num;

if ( ctx_p->synctimeout )

threadinfo_p->expiretime = threadinfo_p->starttime + ctx_p->synctimeout;

if ( pthread_create ( &threadinfo_p->pthread, NULL, ( void * ( * ) ( void * ) ) so_call_sync_thread, threadinfo_p ) ) {

error ( "Cannot pthread_create()." );

return errno;

}

debug ( 3, "thread %p", threadinfo_p->pthread );

return 0;

#endif

}

static inline int so_call_rsync_finished ( ctx_t *ctx_p, const char *inclistfile, const char *exclistfile )

{

int ret0, ret1;

debug ( 5, "" );

if ( ctx_p->flags[DONTUNLINK] )

return 0;

if ( inclistfile == NULL ) {

error ( "inclistfile == NULL." );

return EINVAL;

}

debug ( 3, "unlink()-ing \"%s\"", inclistfile );

ret0 = unlink ( inclistfile );

if ( ctx_p->flags[RSYNCPREFERINCLUDE] )

return ret0;

if ( exclistfile == NULL ) {

error ( "exclistfile == NULL." );

return EINVAL;

}

debug ( 3, "unlink()-ing \"%s\"", exclistfile );

ret1 = unlink ( exclistfile );

return ret0 == 0 ? ret1 : ret0;

}

#ifdef THREADING_SUPPORT

int so_call_rsync_thread ( threadinfo_t *threadinfo_p )

{

debug ( 3, "thread_num == %i; threadinfo_p == %p; i_p->pthread %p; thread %p",

threadinfo_p->thread_num, threadinfo_p, threadinfo_p->pthread, pthread_self() );

ctx_t *ctx_p = threadinfo_p->ctx_p;

char **argv = threadinfo_p->argv;

int err = 0, rc = 0, try_again;

do {

try_again = 0;

threadinfo_p->try_n++;

rc = ctx_p->handler_funct.rsync ( argv[0], argv[1] );

if ( ( err = exitcode_process ( threadinfo_p->ctx_p, rc ) ) ) {

try_again = ( ( !ctx_p->retries ) || ( threadinfo_p->try_n < ctx_p->retries ) ) && ( ctx_p->state != STATE_TERM ) && ( ctx_p->state != STATE_EXIT );

warning ( "Bad exitcode %i (errcode %i). %s.", rc, err, try_again ? "Retrying" : "Give up" );

if ( try_again ) {

debug ( 2, "Sleeping for %u seconds before the retry.", ctx_p->syncdelay );

sleep ( ctx_p->syncdelay );

}

}

} while ( try_again );

if ( err && !ctx_p->flags[IGNOREFAILURES] ) {

error ( "Bad exitcode %i (errcode %i)", rc, err );

threadinfo_p->errcode = err;

}

if ( ( err = so_call_rsync_finished ( ctx_p, argv[0], argv[1] ) ) ) {

exitcode = err; // This's global variable "exitcode"

pthread_kill ( pthread_sighandler, SIGTERM );

}

free ( argv[0] );

free ( argv[1] );

free ( argv );

if ( ( err = thread_exit ( threadinfo_p, rc ) ) ) {

exitcode = err; // This's global variable "exitcode"

pthread_kill ( pthread_sighandler, SIGTERM );

}

return rc;

}

#endif

static inline int so_call_rsync ( ctx_t *ctx_p, indexes_t *indexes_p, const char *inclistfile, const char *exclistfile )

{

debug ( 2, "inclistfile == \"%s\"; exclistfile == \"%s\"", inclistfile, exclistfile );

#ifdef THREADING_SUPPORT

if ( !SHOULD_THREAD ( ctx_p ) ) {

#endif

debug ( 3, "ctx_p->handler_funct.rsync == %p", ctx_p->handler_funct.rsync );

// indexes_p->nonthreaded_syncing_fpath2ei_ht = g_hash_table_dup(indexes_p->fpath2ei_ht, g_str_hash, g_str_equal, free, free, (gpointer(*)(gpointer))strdup, eidup);

indexes_p->nonthreaded_syncing_fpath2ei_ht = indexes_p->fpath2ei_ht;

int rc = 0, err = 0;

int try_n = 0, try_again;

state_t status = STATE_UNKNOWN;

do {

try_again = 0;

try_n++;

alarm ( ctx_p->synctimeout );

rc = ctx_p->handler_funct.rsync ( inclistfile, exclistfile );

alarm ( 0 );

if ( ( err = exitcode_process ( ctx_p, rc ) ) ) {

if ( ( try_n == 1 ) && ( ctx_p->state != STATE_TERM ) && ( ctx_p->state != STATE_EXIT ) ) {

status = ctx_p->state;

ctx_p->state = STATE_SYNCHANDLER_ERR;

main_status_update ( ctx_p );

}

try_again = ( ( !ctx_p->retries ) || ( try_n < ctx_p->retries ) ) && ( ctx_p->state != STATE_TERM ) && ( ctx_p->state != STATE_EXIT );

warning ( "Bad exitcode %i (errcode %i). %s.", rc, err, try_again ? "Retrying" : "Give up" );

if ( try_again ) {

debug ( 2, "Sleeping for %u seconds before the retry.", ctx_p->syncdelay );

sleep ( ctx_p->syncdelay );

}

}

} while ( try_again );

if ( err && !ctx_p->flags[IGNOREFAILURES] ) {

error ( "Bad exitcode %i (errcode %i)", rc, err );

rc = err;

} else if ( ( status != STATE_UNKNOWN ) && ( ctx_p->state != STATE_TERM ) && ( ctx_p->state != STATE_EXIT ) ) {

ctx_p->state = status;

main_status_update ( ctx_p );

}

// g_hash_table_destroy(indexes_p->nonthreaded_syncing_fpath2ei_ht);

indexes_p->nonthreaded_syncing_fpath2ei_ht = NULL;

int ret_cleanup;

if ( ( ret_cleanup = so_call_rsync_finished ( ctx_p, inclistfile, exclistfile ) ) )

return rc ? rc : ret_cleanup;

return rc;

#ifdef THREADING_SUPPORT

}

threadinfo_t *threadinfo_p = thread_new();

if ( threadinfo_p == NULL )

return errno;

threadinfo_p->try_n = 0;

threadinfo_p->callback = NULL;

threadinfo_p->argv = xmalloc ( sizeof ( char * ) * 3 );

threadinfo_p->ctx_p = ctx_p;

threadinfo_p->starttime = time ( NULL );

threadinfo_p->fpath2ei_ht = g_hash_table_dup ( indexes_p->fpath2ei_ht, g_str_hash, g_str_equal, free, free, ( gpointer ( * ) ( gpointer ) ) strdup, eidup );

threadinfo_p->iteration = ctx_p->iteration_num;

threadinfo_p->argv[0] = strdup ( inclistfile );

threadinfo_p->argv[1] = strdup ( exclistfile );

if ( ctx_p->synctimeout )

threadinfo_p->expiretime = threadinfo_p->starttime + ctx_p->synctimeout;

if ( pthread_create ( &threadinfo_p->pthread, NULL, ( void * ( * ) ( void * ) ) so_call_rsync_thread, threadinfo_p ) ) {

error ( "Cannot pthread_create()." );

return errno;

}

debug ( 3, "thread %p", threadinfo_p->pthread );

return 0;

#endif

}

// === SYNC_EXEC() === {

//#define SYNC_EXEC(...) (SHOULD_THREAD(ctx_p) ? sync_exec_thread : sync_exec )(__VA_ARGS__)

#ifdef THREADING_SUPPORT

#define SYNC_EXEC_ARGV(...) (SHOULD_THREAD(ctx_p) ? sync_exec_argv_thread : sync_exec_argv)(__VA_ARGS__)

#else

#define SYNC_EXEC_ARGV(...) sync_exec_argv(__VA_ARGS__)

#endif

#define debug_argv_dump(level, argv)\

if (unlikely(ctx_p->flags[DEBUG] >= level))\

argv_dump(level, argv)

static inline void argv_dump ( int debug_level, char **argv )

{

#ifdef _DEBUG_FORCE

debug ( 19, "(%u, %p)", debug_level, argv );

#endif

char **argv_p = argv;

while ( *argv_p != NULL ) {

debug ( debug_level, "%p: \"%s\"", *argv_p, *argv_p );

argv_p++;

}

return;

}

#define _sync_exec_getargv(argv, firstarg, COPYARG) {\

va_list arglist;\

va_start(arglist, firstarg);\

\

int i = 0;\

do {\

char *arg;\

if(i >= MAXARGUMENTS) {\

error("Too many arguments (%i >= %i).", i, MAXARGUMENTS);\

return ENOMEM;\

}\

arg = (char *)va_arg(arglist, const char *const);\

argv[i] = arg!=NULL ? COPYARG : NULL;\

} while(argv[i++] != NULL);\

va_end(arglist);\

}

char *sync_path_rel2abs ( ctx_t *ctx_p, const char *path_rel, ssize_t path_rel_len, size_t *path_abs_len_p, char *path_abs_oldptr )

{

if ( path_rel == NULL )

return NULL;

if ( path_rel_len == -1 )

path_rel_len = strlen ( path_rel );

char *path_abs;

size_t watchdirlen =

( ctx_p->watchdir == ctx_p->watchdirwslash ) ? 0 : ctx_p->watchdirlen;

// if [watchdir == "/"] ? 0 : watchdir.length()

size_t path_abs_len = path_rel_len + watchdirlen + 1;