replica.c

   1 /* Copyright (C) 2010 Red Hat, Inc.
   2
   3    This program is free software: you can redistribute it and/or modify
   4    it under the terms of the GNU General Public License as published by
   5    the Free Software Foundation, either version 3 of the License, or
   6    (at your option) any later version.
   7
   8    This program is distributed in the hope that it will be useful,
   9    but WITHOUT ANY WARRANTY; without even the implied warranty of
  10    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11    GNU General Public License for more details.
  12
  13    You should have received a copy of the GNU General Public License
  14    along with this program.  If not, see <http://www.gnu.org/licenses/>.  */
  15
  16 #include <config.h>
  17
  18 #include <errno.h>
  19 #include <error.h>
  20 #include <fcntl.h>
  21 #include <poll.h>
  22 #include <pthread.h>
  23 #include <semaphore.h>
  24 #include <stdint.h>
  25 #include <stdio.h>
  26 #include <stdlib.h>
  27 #include <string.h>
  28 #include <unistd.h>
  29
  30 #include <hstor.h>
  31 #include <microhttpd.h> /* for HTTP status values */
  32
  33 #include "iwh.h"
  34 #include "setup.h"
  35 #include "query.h"
  36 #include "meta.h"
  37 #include "replica.h"
  38
  39 /* Sizes for internal string buffers. */
  40 #define ADDR_SIZE       1024
  41 #define SVC_ACC_SIZE    128
  42 #define HEADER_SIZE     64
  43
  44 typedef enum {
  45         REPL_PUT,               /* store an object */
  46         REPL_ODELETE,           /* delete an object */
  47         REPL_BCREATE,           /* create a bucket */
  48         /* TBD: bucket deletion, others? */
  49 } repl_t;
  50
  51 typedef struct _repl_item {
  52         struct _repl_item       *next;
  53         repl_t                   type;
  54         char                    *path;
  55         provider_t              *server;
  56         size_t                   size;
  57         int                      pipes[2];
  58         my_state                *ms;
  59 } repl_item;
  60
  61 typedef struct {
  62         char            *cur_bucket;
  63         char            *cur_key;
  64         provider_t      *cur_server;
  65 } query_ctx_t;
  66
  67 static repl_item        *queue_head     = NULL;
  68 static repl_item        *queue_tail     = NULL;
  69 static pthread_mutex_t   queue_lock;
  70 static sem_t             queue_sema;
  71 static volatile gint     rep_count      = 0;
  72
  73 static void *
  74 proxy_repl_prod (void *ctx)
  75 {
  76         repl_item               *item   = ctx;
  77         backend_thunk_t          thunk;
  78         void                    *result;
  79
  80         thunk.parent = item->ms;
  81         thunk.prov = main_prov;
  82
  83         result = thunk.prov->func_tbl->get_child_func(&thunk);
  84         return result;
  85 }
  86
  87 static void *
  88 proxy_repl_cons (void *ctx)
  89 {
  90         repl_item               *item   = ctx;
  91         my_state                *ms     = item->ms;
  92         pipe_private            *pp;
  93
  94         pp = pipe_init_private(&ms->pipe);
  95         if (!pp) {
  96                 pipe_cons_siginit(&ms->pipe,-1);
  97                 return THREAD_FAILED;
  98         }
  99
 100         pp->prov = item->server;
 101         ms->be_flags = 0;
 102
 103         return item->server->func_tbl->put_child_func(pp);
 104 }
 105
 106 static void
 107 repl_worker_del (const repl_item *item)
 108 {
 109         char    *bucket;
 110         char    *key;
 111         int      rc;
 112
 113         bucket = strdup(item->path);
 114         if (!bucket) {
 115                 error(0,errno,"ran out of memory replicating delete for %s",
 116                         item->path);
 117                 return;
 118         }
 119
 120         key = strchr(bucket,'/');
 121         if (!key) {
 122                 error(0,0,"invalid path replicating delete for %s",item->path);
 123                 free(bucket);
 124                 return;
 125         }
 126         ++key;
 127
 128         rc = item->server->func_tbl->delete_func(item->server,
 129                 bucket, key, item->path);
 130         if (rc != MHD_HTTP_OK) {
 131                 error(0,0,"got status %d replicating delete for %s",
 132                         rc, item->path);
 133         }
 134         free(bucket);
 135
 136         DPRINTF("finished replicating delete for %s, rc = %d\n",item->path,rc);
 137 }
 138
 139 static void
 140 repl_worker_bcreate (repl_item *item)
 141 {
 142         int      rc;
 143
 144         rc = item->server->func_tbl->bcreate_func(item->server,item->path);
 145         if (rc != MHD_HTTP_OK) {
 146                 error(0,0,"got status %d replicating bcreate for %s",
 147                         rc, item->path);
 148         }
 149
 150         DPRINTF("finished replicating bcreate for %s, rc = %d\n",item->path,rc);
 151 }
 152
 153 /* Use this to diagnose failed thread creation.  */
 154 #define xpthread_create(thread, start_routine, item, msg)               \
 155   do {                                                                  \
 156     int err = pthread_create (thread, NULL, start_routine, item);       \
 157     if (err) {                                                          \
 158       error (0, err, msg);                                              \
 159       return NULL;                                                      \
 160     }                                                                   \
 161   } while (0)
 162
 163 static void *
 164 repl_worker (void *notused ATTRIBUTE_UNUSED)
 165 {
 166         repl_item       *item;
 167         pthread_t        cons;
 168         pthread_t        prod;
 169         my_state        *ms;
 170
 171         for (;;) {
 172                 sem_wait(&queue_sema);
 173                 pthread_mutex_lock(&queue_lock);
 174                 item = queue_head;
 175                 queue_head = item->next;
 176                 if (!queue_head) {
 177                         queue_tail = NULL;
 178                 }
 179                 pthread_mutex_unlock(&queue_lock);
 180
 181                 /*
 182                  * Do a full initialization here, not just in the rest.  It's
 183                  * necessary in the oddball case where we're re-replicating as
 184                  * a result of an attribute/policy change, and it's not harmful
 185                  * in the normal case where we're actually storing a new file.
 186                  */
 187                 ms = item->ms;
 188                 pipe_init_shared(&ms->pipe,ms,1);
 189                 switch (item->type) {
 190                 case REPL_PUT:
 191                         if (pipe(item->pipes) >= 0) {
 192                                 xpthread_create(&prod,proxy_repl_prod,item,
 193                                             "failed to start producer thread");
 194                                 xpthread_create(&cons,proxy_repl_cons,item,
 195                                             "failed to start consumer thread");
 196                                 pthread_join(prod,NULL);
 197                                 pthread_join(cons,NULL);
 198                         }
 199                         else {
 200                                 error(0,errno,"pipe");
 201                         }
 202                         break;
 203                 case REPL_ODELETE:
 204                         repl_worker_del(item);
 205                         break;
 206                 case REPL_BCREATE:
 207                         repl_worker_bcreate(item);
 208                         break;
 209                 default:
 210                         error(0,0,"bad repl type %d (url=%s) skipped",
 211                                 item->type, item->path);
 212                 }
 213                 free_ms(item->ms);
 214                 free(item->path);
 215                 free(item);
 216                 /* No atomic dec without test?  Lame. */
 217                 (void)g_atomic_int_dec_and_test(&rep_count);
 218         }
 219 }
 220
 221 void
 222 repl_init (void)
 223 {
 224         pthread_t       tid;
 225
 226         sem_init(&queue_sema,0,0);
 227         pthread_mutex_init(&queue_lock,NULL);
 228         pthread_create(&tid,NULL,repl_worker,NULL);
 229 }
 230
 231 static const char *
 232 repl_oget (void *ctx, const char *id)
 233 {
 234         query_ctx_t     *qctx = ctx;
 235         char            *cur_value = NULL;
 236
 237         (void)meta_get_value(qctx->cur_bucket,qctx->cur_key,id,&cur_value);
 238
 239         return cur_value;
 240 }
 241
 242 static const char *
 243 repl_sget (void *ctx, const char *id)
 244 {
 245         query_ctx_t     *qctx   = ctx;
 246         provider_t      *prov   = qctx->cur_server;
 247
 248         if (!strcmp(id,"name")) {
 249                 return prov->name;
 250         }
 251         if (!strcmp(id,"type")) {
 252                 return prov->type;
 253         }
 254         if (!strcmp(id,"host")) {
 255                 return prov->host;
 256         }
 257         if (!strcmp(id,"key")) {
 258                 return prov->username;
 259         }
 260         if (!strcmp(id,"secret")) {
 261                 return prov->password;
 262         }
 263         if (!strcmp(id,"path")) {
 264                 return prov->path;
 265         }
 266
 267         return g_hash_table_lookup(prov->attrs,id);
 268 }
 269
 270 void
 271 replicate (const char *url, size_t size, const char *policy, my_state *ms)
 272 {
 273         repl_item       *item;
 274         value_t         *expr;
 275         int              res;
 276         char            *url2;
 277         char            *stctx;
 278         query_ctx_t      qctx;
 279         getter_t         oget;
 280         getter_t         sget;
 281         GHashTableIter   iter;
 282         gpointer         key;
 283         gpointer         value;
 284         provider_t      *prov;
 285
 286         url2 = strdup(url);
 287         if (!url2) {
 288                 error(0,0,"could not parse url %s",url);
 289                 return;
 290         }
 291         qctx.cur_bucket = strtok_r(url2,"/",&stctx);
 292         qctx.cur_key = strtok_r(NULL,"/",&stctx);
 293
 294         if (!size) {
 295                 size = meta_get_size(qctx.cur_bucket,qctx.cur_key);
 296                 DPRINTF("fetched size %zu for %s\n",size,url);
 297         }
 298
 299         if (policy) {
 300                 DPRINTF("--- policy = %s\n",policy);
 301                 expr = parse(policy);
 302         }
 303         else {
 304                 expr = NULL;
 305         }
 306
 307         oget.func = repl_oget;
 308         oget.ctx = &qctx;
 309         sget.func = repl_sget;
 310         sget.ctx = &qctx;
 311
 312         init_prov_iter(&iter);
 313         while (g_hash_table_iter_next(&iter,&key,&value)) {
 314                 if (!strcmp(key,me)) {
 315                         continue;
 316                 }
 317                 prov = (provider_t *)value;
 318                 if (expr) {
 319                         qctx.cur_server = prov;
 320                         res = eval(expr,&oget,&sget);
 321                 }
 322                 else {
 323                         res = 0;
 324                 }
 325                 if (res <= 0) {
 326                         DPRINTF("skipping %s for %s\n",prov->name,url);
 327                         continue;
 328                 }
 329                 DPRINTF("REPLICATING %s to %s\n",url,prov->name);
 330                 item = malloc(sizeof(*item));
 331                 if (!item) {
 332                         error(0,errno,"could not create repl_item for %s",
 333                               url);
 334                         break;
 335                 }
 336                 item->type = REPL_PUT;
 337                 item->path = strdup(url);
 338                 if (!item->path) {
 339                         error(0,errno,"could not create repl_item for %s",
 340                               url);
 341                         break;
 342                 }
 343                 item->server = prov;
 344                 item->size = size;
 345                 item->ms = ms;
 346                 g_atomic_int_inc(&ms->refcnt);
 347                 pthread_mutex_lock(&queue_lock);
 348                 if (queue_tail) {
 349                         item->next = queue_tail->next;
 350                         queue_tail->next = item;
 351                 }
 352                 else {
 353                         item->next = NULL;
 354                         queue_head = item;
 355                 }
 356                 queue_tail = item;
 357                 pthread_mutex_unlock(&queue_lock);
 358                 g_atomic_int_inc(&rep_count);
 359                 sem_post(&queue_sema);
 360         }
 361
 362         if (expr) {
 363                 free_value(expr);
 364         }
 365         free(url2);
 366 }
 367
 368 static void
 369 replicate_namespace_action (const char *name, repl_t action, my_state *ms)
 370 {
 371         repl_item       *item;
 372         GHashTableIter   iter;
 373         gpointer         key;
 374         gpointer         value;
 375
 376         init_prov_iter(&iter);
 377         while (g_hash_table_iter_next(&iter,&key,&value)) {
 378                 if (!strcmp(key,me)) {
 379                         continue;
 380                 }
 381                 DPRINTF("replicating %s(%s) on %s\n",
 382                         (action == REPL_ODELETE ? "delete" : "create"),
 383                         name,
 384                         ((provider_t *)value)->name);
 385                 item = malloc(sizeof(*item));
 386                 if (!item) {
 387                         error(0,errno,"could not create repl_item for %s",
 388                               name);
 389                         return;
 390                 }
 391                 item->type = action;
 392                 item->path = strdup(name);
 393                 if (!item->path) {
 394                         free(item);
 395                         return;
 396                 }
 397                 item->server = (provider_t *)value;
 398                 item->ms = ms;
 399                 g_atomic_int_inc(&ms->refcnt);
 400                 pthread_mutex_lock(&queue_lock);
 401                 if (queue_tail) {
 402                         item->next = queue_tail->next;
 403                         queue_tail->next = item;
 404                 }
 405                 else {
 406                         item->next = NULL;
 407                         queue_head = item;
 408                 }
 409                 queue_tail = item;
 410                 pthread_mutex_unlock(&queue_lock);
 411                 g_atomic_int_inc(&rep_count);
 412                 sem_post(&queue_sema);
 413         }
 414 }
 415
 416 void
 417 replicate_delete (const char *name, my_state *ms)
 418 {
 419         replicate_namespace_action(name,REPL_ODELETE,ms);
 420 }
 421
 422 void
 423 replicate_bcreate (const char *name, my_state *ms)
 424 {
 425         replicate_namespace_action(name,REPL_BCREATE,ms);
 426 }
 427
 428 /* Part of our API to the query module. */
 429 char *
 430 follow_link (char *object, const char *key)
 431 {
 432         char    *slash;
 433         char    *value  = NULL;
 434
 435         slash = strchr(object,'/');
 436         if (!slash) {
 437                 return NULL;
 438         }
 439
 440         *(slash++) = '\0';
 441         (void)meta_get_value(object,slash,key,&value);
 442         *(--slash) = '/';
 443
 444         DPRINTF("%s: %s:%s => %s\n",__func__,object,key,value);
 445         return value;
 446 }
 447
 448 int
 449 get_rep_count (void)
 450 {
 451         return g_atomic_int_get(&rep_count);
 452 }