mon_gio.c

Go to the documentation of this file.

/*
    clsync - file tree sync utility based on inotify/kqueue/bsm/gio
 
    Copyright (C) 2013-2014 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/>.
 */
 
// The "queue" is actually "stack" in this code. It's a lack of design of this code.
 
#include "common.h"
#include "error.h"
#include "sync.h"
#include "indexes.h"
#include "privileged.h"
 
#include <pthread.h>
#include <gio/gio.h>
 
#include "mon_gio.h"
 
struct filemondata {
    ctx_t       *ctx_p;
    GFile       *file;
    GFileMonitor    *filemon;
    gulong       handle_id;
};
typedef struct filemondata filemondata_t;
 
struct event {
    char *path;
    gulong            handle_id;
    GFileMonitorEvent event_id;
    eventobjtype_t    objtype_event;
    eventobjtype_t    objtype_old;
    eventobjtype_t    objtype_new;
};
typedef struct event event_t;
 
GHashTable *mondirs_ht;
pthread_spinlock_t queue_lock;
pthread_mutex_t    gio_mutex_prefetcher = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t     gio_cond_gotevent    = PTHREAD_COND_INITIALIZER;
event_t *queue = NULL;
int queue_length;
int queue_alloc;
 
static inline void event_free ( event_t *ev )
{
    free ( ev->path );
    return;
}
 
static inline int event_push ( char *path, gulong handle_id, GFileMonitorEvent event, eventobjtype_t objtype_event, eventobjtype_t objtype_old, eventobjtype_t objtype_new )
{
    event_t *ev;
    debug ( 30, "pthread_spin_lock(&queue_lock);" );
    pthread_spin_lock ( &queue_lock );
 
    if ( queue_length >= queue_alloc ) {
        queue_alloc += ALLOC_PORTION;
        critical_on ( queue_alloc >= GIO_QUEUE_LENGTH_MAX );
        queue = xrealloc ( queue, queue_alloc * sizeof ( *queue ) );
    }
 
    ev = &queue[queue_length++];
    ev->path          = path;
    ev->event_id      = event;
    ev->handle_id     = handle_id;
    ev->objtype_event = objtype_event;
    ev->objtype_old   = objtype_old;
    ev->objtype_new   = objtype_new;
    debug ( 30, "pthread_spin_unlock(&queue_lock);" );
    pthread_spin_unlock ( &queue_lock );
    return 0;
}
 
static inline event_t *event_pop()
{
    static event_t ev;
    debug ( 30, "pthread_spin_lock(&queue_lock);" );
    pthread_spin_lock ( &queue_lock );
    critical_on ( !queue_length );
    memcpy ( &ev, &queue[--queue_length], sizeof ( ev ) );
    debug ( 30, "pthread_spin_unlock(&queue_lock);" );
    pthread_spin_unlock ( &queue_lock );
    return &ev;
}
 
static void dir_gotevent (
    GFileMonitor        *filemon,
    GFile           *file,
    GFile           *file_other,
    GFileMonitorEvent    event,
    gpointer         arg
)
{
    eventobjtype_t objtype_old, objtype_new, objtype;
    filemondata_t *fmdat    = arg;
    ctx_t         *ctx_p    = fmdat->ctx_p;
    GFileType      filetype = g_file_query_file_type ( file, G_FILE_QUERY_INFO_NOFOLLOW_SYMLINKS, NULL );
    debug ( 10, "%p %p %p %i %p %i", filemon, file, file_other, event, arg, filetype );
    char *path_full, *path_rel = NULL;
 
    switch ( event ) {
        case G_FILE_MONITOR_EVENT_DELETED:
        case G_FILE_MONITOR_EVENT_CREATED:
        case G_FILE_MONITOR_EVENT_CHANGED:
        case G_FILE_MONITOR_EVENT_ATTRIBUTE_CHANGED:
            path_full = g_file_get_path ( file );
            path_rel  = strdup ( &path_full[ctx_p->watchdirlen + 1] );
            g_free ( path_full );
            debug ( 9, "Got event %i for \"%s\" (%i)", event, path_rel, filetype );
            break;
 
        default:
            break;
    }
 
    switch ( filetype ) {
        case G_FILE_TYPE_DIRECTORY:
            objtype = EOT_DIR;
            break;
 
        default:
            objtype = EOT_FILE;
            break;
    }
 
    switch ( event ) {
        case G_FILE_MONITOR_EVENT_DELETED:
            objtype_old = objtype;
            objtype_new = EOT_DOESNTEXIST;
            break;
 
        case G_FILE_MONITOR_EVENT_CREATED:
            objtype_old = EOT_DOESNTEXIST;
            objtype_new = objtype;
            break;
 
        default:
            objtype_old = objtype;
            objtype_new = objtype;
            break;
    }
 
    switch ( event ) {
        case G_FILE_MONITOR_EVENT_DELETED:
            debug ( 20, "g_hash_table_remove(mondirs_ht, \"%s\")", path_rel );
            g_hash_table_remove ( mondirs_ht, path_rel );
 
        case G_FILE_MONITOR_EVENT_CREATED:
        case G_FILE_MONITOR_EVENT_CHANGED:
        case G_FILE_MONITOR_EVENT_ATTRIBUTE_CHANGED:
#ifdef PARANOID
            critical_on ( path_rel == NULL );
#endif
            debug ( 20, "event_push(\"%s\", %i, %i, %i, %i, %i)", path_rel, fmdat->handle_id, event, objtype, objtype_old, objtype_new );
            critical_on ( event_push ( path_rel, fmdat->handle_id, event, objtype, objtype_old, objtype_new ) );
            break;
 
        default:
            break;
    }
 
    return;
}
 
int gio_add_watch_dir ( ctx_t *ctx_p, indexes_t *indexes_p, const char *const accpath )
{
    ( void ) indexes_p;
    filemondata_t *fmdat;
    GError        *error = NULL;
    debug ( 3, "\"%s\"", accpath );
#ifdef PARANOID
    fmdat = g_hash_table_lookup ( mondirs_ht, accpath );
 
    if ( fmdat != NULL ) {
        errno = EADDRINUSE;
        warning ( "Directory \"%s\" is already monitored.", accpath );
        return -1;
    }
 
#endif
    fmdat = xmalloc ( sizeof ( *fmdat ) );
    fmdat->ctx_p     = ctx_p;
    fmdat->file      = g_file_new_for_path ( accpath );
    fmdat->filemon   = g_file_monitor_directory ( fmdat->file, 0, NULL, &error );
    fmdat->handle_id = g_signal_connect ( fmdat->filemon, "changed", G_CALLBACK ( dir_gotevent ), fmdat );
    g_hash_table_replace ( mondirs_ht, strdup ( accpath ), fmdat );
    return fmdat->handle_id;
}
 
int cancel_g_iteration_stop;
pthread_t thread_g_iteration_stop;
void *g_iteration_stop ( void *_timeout_p )
{
    struct timeval *timeout_p = _timeout_p;
    struct timeval tv_abs, timeout_abs;
    struct timespec ts_abs;
    debug ( 10, "{%u, %u}", timeout_p->tv_sec, timeout_p->tv_usec );
    critical_on ( pthread_mutex_lock ( &gio_mutex_prefetcher ) );
 
    if ( cancel_g_iteration_stop ) {
        critical_on ( pthread_mutex_unlock ( &gio_mutex_prefetcher ) );
        return NULL;
    }
 
#define INFINITETIME (3600 * 24 * 365 * 10) /* ~10 years */
 
    if ( timeout_p->tv_sec > INFINITETIME )
        timeout_p->tv_sec = INFINITETIME;
 
#undef INFINITETIME
    gettimeofday ( &tv_abs, NULL );
    timeradd ( &tv_abs, timeout_p, &timeout_abs );
    ts_abs.tv_sec  = timeout_abs.tv_sec;
    ts_abs.tv_nsec = timeout_abs.tv_usec * 1000;
    debug ( 10, "{%u, %u}", ts_abs.tv_sec, ts_abs.tv_nsec );
 
    switch ( ( errno = pthread_cond_timedwait ( &gio_cond_gotevent, &gio_mutex_prefetcher, &ts_abs ) ) ) {
        case 0:
        case ETIMEDOUT:
            break;
 
        default:
            critical ( "Got error while pthread_cond_timedwait(&gio_cond_gotevent, &gio_mutex_prefetcher, &ts_abs)" );
    }
 
    g_main_context_wakeup ( NULL );
    pthread_mutex_unlock ( &gio_mutex_prefetcher );
    debug ( 10, "return" );
    return NULL;
}
 
static inline int gio_wait_now ( ctx_t *ctx_p, struct indexes *indexes_p, struct timeval *tv_p )
{
    ( void ) ctx_p;
    ( void ) indexes_p;
    void *ret;
    int result;
    debug ( 3, "(ctx_p, indexes_p, %p {%u, %u})", tv_p, tv_p == NULL ? -1 : tv_p->tv_sec, tv_p == NULL ? 0 : tv_p->tv_usec );
#ifdef PARANOID
    critical_on ( tv_p == NULL );
#endif
 
    if ( queue_length ) {
        debug ( 9, "already: queue_length == %i", queue_length );
        return queue_length;
    }
 
    if ( tv_p->tv_sec == 0 && tv_p->tv_usec == 0 ) {
        g_main_context_iteration ( NULL, FALSE );
        debug ( 9, "nowait: queue_length == %i", queue_length );
        return queue_length;
    }
 
    cancel_g_iteration_stop = 0;
    pthread_create ( &thread_g_iteration_stop, NULL, g_iteration_stop, tv_p );
    /*
        debug(30, "pthread_spin_unlock(&queue_lock);");
        pthread_spin_unlock(&queue_lock);
        debug(30 , "g_main_context_iteration(NULL, FALSE);");
        result  = g_main_context_iteration(NULL, FALSE);
        debug(30, "pthread_spin_lock(&queue_lock);");
        pthread_spin_lock(&queue_lock);
 
        if (queue_length) {
            debug(9, "already2: queue_length == %i", queue_length);
            return queue_length;
        }
    */
    debug_call  ( 40, pthread_spin_unlock ( &queue_lock ) );
    debug ( 20, "g_main_context_iteration(NULL, TRUE); queue_length == %i", queue_length );
    result  = g_main_context_iteration ( NULL, TRUE );
    debug ( 10, "g_main_context_iteration() -> %i", result );
    debug_call  ( 40, pthread_spin_lock ( &queue_lock ) );
    critical_on ( pthread_mutex_lock ( &gio_mutex_prefetcher ) );
    cancel_g_iteration_stop = 1;
    critical_on ( pthread_mutex_unlock ( &gio_mutex_prefetcher ) );
    critical_on ( pthread_cond_broadcast ( &gio_cond_gotevent ) );
    critical_on ( pthread_join ( thread_g_iteration_stop, &ret ) );
    debug ( 9, "queue_length == %i", queue_length );
    return queue_length;
}
int gio_wait ( ctx_t *ctx_p, struct indexes *indexes_p, struct timeval *tv_p )
{
    ( void ) ctx_p;
    int ret;
    debug ( 30, "pthread_spin_lock(&queue_lock);" );
    debug_call ( 40, pthread_spin_lock ( &queue_lock ) );
    ret = gio_wait_now ( ctx_p, indexes_p, tv_p );
    debug ( 30, "pthread_spin_unlock(&queue_lock);" );
    debug_call ( 40, pthread_spin_unlock ( &queue_lock ) );
    return ret;
}
 
int gio_handle ( ctx_t *ctx_p, indexes_t *indexes_p )
{
    static struct timeval tv = {0};
    int count;
    char   *path_full    = NULL;
    size_t  path_full_len    = 0;
    count = 0;
 
    while ( gio_wait ( ctx_p, indexes_p, &tv ) ) {
        event_t *ev = event_pop();
        stat64_t lst, *lst_p;
        mode_t st_mode;
        size_t st_size;
 
        if ( ( ev->objtype_new == EOT_DOESNTEXIST ) || ( ctx_p->flags[CANCEL_SYSCALLS]&CSC_MON_STAT ) || lstat64 ( ev->path, &lst ) ) {
            debug ( 2, "Cannot lstat64(\"%s\", lst). Seems, that the object had been deleted (%i) or option \"--cancel-syscalls mon_stat\" (%i) is set.", ev->path, ev->objtype_new == EOT_DOESNTEXIST, ctx_p->flags[CANCEL_SYSCALLS]&CSC_MON_STAT );
            st_mode = ( ev->objtype_event == EOT_DIR ? S_IFDIR : S_IFREG );
            st_size = 0;
            lst_p = NULL;
        } else {
            st_mode = lst.st_mode;
            st_size = lst.st_size;
            lst_p = &lst;
        }
 
        if ( sync_prequeue_loadmark ( 1, ctx_p, indexes_p, NULL, ev->path, lst_p, ev->objtype_old, ev->objtype_new, ev->event_id, ev->handle_id, st_mode, st_size, &path_full, &path_full_len, NULL ) ) {
            event_free ( ev );
            count = -1;
            break;
        }
 
        event_free ( ev );
        count++;
    }
 
    // Globally queueing captured events:
    // Moving events from local queue to global ones
    sync_prequeue_unload ( ctx_p, indexes_p );
    free ( path_full );
#ifdef VERYPARANOID
    path_full     = NULL;
    path_full_len = 0;
#endif
    return count;
}
 
void free_filemondat ( void *_fmdat )
{
    filemondata_t *fmdat = _fmdat;
    g_signal_handler_disconnect ( fmdat->file, fmdat->handle_id );
    free ( fmdat->file );
    free ( fmdat->filemon );
    free ( fmdat );
    return;
}
 
GMainLoop *gio_loop = NULL;
int gio_init ( ctx_t *ctx_p )
{
    ( void ) ctx_p;
    queue_length = 0;
    queue_alloc  = 0;
    pthread_mutex_init ( &gio_mutex_prefetcher, NULL );
    pthread_cond_init  ( &gio_cond_gotevent,    NULL );
    pthread_spin_init  ( &queue_lock,           PTHREAD_PROCESS_SHARED );
    mondirs_ht = g_hash_table_new_full ( g_str_hash, g_str_equal, free, free_filemondat );
    gio_loop   = g_main_loop_new ( NULL, TRUE );
    g_main_context_iteration ( NULL, FALSE );
    return 0;
}
 
int gio_deinit ( ctx_t *ctx_p )
{
    ( void ) ctx_p;
    /*
        g_main_loop_quit(gio_loop);
        g_hash_table_destroy  (mondirs_ht);
        pthread_spin_destroy  (&queue_lock);
        pthread_cond_destroy  (&gio_cond_gotevent);
        pthread_mutex_destroy (&gio_mutex_prefetcher);
    */
    return 0;
}