1 /* Copyright (C) 2010 Red Hat, Inc.
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.
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.
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/>. */
23 #include <semaphore.h>
31 #include <microhttpd.h> /* for HTTP status values */
39 /* Sizes for internal string buffers. */
40 #define ADDR_SIZE 1024
41 #define SVC_ACC_SIZE 128
42 #define HEADER_SIZE 64
45 REPL_PUT
, /* store an object */
46 REPL_ODELETE
, /* delete an object */
47 REPL_BCREATE
, /* create a bucket */
48 /* TBD: bucket deletion, others? */
51 typedef struct _repl_item
{
52 struct _repl_item
*next
;
64 provider_t
*cur_server
;
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;
74 proxy_repl_prod (void *ctx
)
76 repl_item
*item
= ctx
;
77 backend_thunk_t thunk
;
80 thunk
.parent
= item
->ms
;
81 thunk
.prov
= main_prov
;
83 result
= thunk
.prov
->func_tbl
->get_child_func(&thunk
);
88 proxy_repl_cons (void *ctx
)
90 repl_item
*item
= ctx
;
91 my_state
*ms
= item
->ms
;
94 pp
= pipe_init_private(&ms
->pipe
);
96 pipe_cons_siginit(&ms
->pipe
,-1);
100 pp
->prov
= item
->server
;
103 return item
->server
->func_tbl
->put_child_func(pp
);
107 repl_worker_del (const repl_item
*item
)
113 bucket
= strdup(item
->path
);
115 error(0,errno
,"ran out of memory replicating delete for %s",
120 key
= strchr(bucket
,'/');
122 error(0,0,"invalid path replicating delete for %s",item
->path
);
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",
136 DPRINTF("finished replicating delete for %s, rc = %d\n",item
->path
,rc
);
140 repl_worker_bcreate (repl_item
*item
)
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",
150 DPRINTF("finished replicating bcreate for %s, rc = %d\n",item
->path
,rc
);
153 /* Use this to diagnose failed thread creation. */
154 #define xpthread_create(thread, start_routine, item, msg) \
156 int err = pthread_create (thread, NULL, start_routine, item); \
158 error (0, err, msg); \
164 repl_worker (void *notused ATTRIBUTE_UNUSED
)
172 sem_wait(&queue_sema
);
173 pthread_mutex_lock(&queue_lock
);
175 queue_head
= item
->next
;
179 pthread_mutex_unlock(&queue_lock
);
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.
188 pipe_init_shared(&ms
->pipe
,ms
,1);
189 switch (item
->type
) {
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
);
200 error(0,errno
,"pipe");
204 repl_worker_del(item
);
207 repl_worker_bcreate(item
);
210 error(0,0,"bad repl type %d (url=%s) skipped",
211 item
->type
, item
->path
);
216 /* No atomic dec without test? Lame. */
217 (void)g_atomic_int_dec_and_test(&rep_count
);
226 sem_init(&queue_sema
,0,0);
227 pthread_mutex_init(&queue_lock
,NULL
);
228 pthread_create(&tid
,NULL
,repl_worker
,NULL
);
232 repl_oget (void *ctx
, const char *id
)
234 query_ctx_t
*qctx
= ctx
;
235 char *cur_value
= NULL
;
237 (void)meta_get_value(qctx
->cur_bucket
,qctx
->cur_key
,id
,&cur_value
);
243 repl_sget (void *ctx
, const char *id
)
245 query_ctx_t
*qctx
= ctx
;
246 provider_t
*prov
= qctx
->cur_server
;
248 if (!strcmp(id
,"name")) {
251 if (!strcmp(id
,"type")) {
254 if (!strcmp(id
,"host")) {
257 if (!strcmp(id
,"key")) {
258 return prov
->username
;
260 if (!strcmp(id
,"secret")) {
261 return prov
->password
;
263 if (!strcmp(id
,"path")) {
267 return g_hash_table_lookup(prov
->attrs
,id
);
271 replicate (const char *url
, size_t size
, const char *policy
, my_state
*ms
)
288 error(0,0,"could not parse url %s",url
);
291 qctx
.cur_bucket
= strtok_r(url2
,"/",&stctx
);
292 qctx
.cur_key
= strtok_r(NULL
,"/",&stctx
);
295 size
= meta_get_size(qctx
.cur_bucket
,qctx
.cur_key
);
296 DPRINTF("fetched size %zu for %s\n",size
,url
);
300 DPRINTF("--- policy = %s\n",policy
);
301 expr
= parse(policy
);
307 oget
.func
= repl_oget
;
309 sget
.func
= repl_sget
;
312 init_prov_iter(&iter
);
313 while (g_hash_table_iter_next(&iter
,&key
,&value
)) {
314 if (!strcmp(key
,me
)) {
317 prov
= (provider_t
*)value
;
319 qctx
.cur_server
= prov
;
320 res
= eval(expr
,&oget
,&sget
);
326 DPRINTF("skipping %s for %s\n",prov
->name
,url
);
329 DPRINTF("REPLICATING %s to %s\n",url
,prov
->name
);
330 item
= malloc(sizeof(*item
));
332 error(0,errno
,"could not create repl_item for %s",
336 item
->type
= REPL_PUT
;
337 item
->path
= strdup(url
);
339 error(0,errno
,"could not create repl_item for %s",
346 g_atomic_int_inc(&ms
->refcnt
);
347 pthread_mutex_lock(&queue_lock
);
349 item
->next
= queue_tail
->next
;
350 queue_tail
->next
= item
;
357 pthread_mutex_unlock(&queue_lock
);
358 g_atomic_int_inc(&rep_count
);
359 sem_post(&queue_sema
);
369 replicate_namespace_action (const char *name
, repl_t action
, my_state
*ms
)
376 init_prov_iter(&iter
);
377 while (g_hash_table_iter_next(&iter
,&key
,&value
)) {
378 if (!strcmp(key
,me
)) {
381 DPRINTF("replicating %s(%s) on %s\n",
382 (action
== REPL_ODELETE
? "delete" : "create"),
384 ((provider_t
*)value
)->name
);
385 item
= malloc(sizeof(*item
));
387 error(0,errno
,"could not create repl_item for %s",
392 item
->path
= strdup(name
);
397 item
->server
= (provider_t
*)value
;
399 g_atomic_int_inc(&ms
->refcnt
);
400 pthread_mutex_lock(&queue_lock
);
402 item
->next
= queue_tail
->next
;
403 queue_tail
->next
= item
;
410 pthread_mutex_unlock(&queue_lock
);
411 g_atomic_int_inc(&rep_count
);
412 sem_post(&queue_sema
);
417 replicate_delete (const char *name
, my_state
*ms
)
419 replicate_namespace_action(name
,REPL_ODELETE
,ms
);
423 replicate_bcreate (const char *name
, my_state
*ms
)
425 replicate_namespace_action(name
,REPL_BCREATE
,ms
);
428 /* Part of our API to the query module. */
430 follow_link (char *object
, const char *key
)
435 slash
= strchr(object
,'/');
441 (void)meta_get_value(object
,slash
,key
,&value
);
444 DPRINTF("%s: %s:%s => %s\n",__func__
,object
,key
,value
);
451 return g_atomic_int_get(&rep_count
);