6 #include <rados/librados.h>
7 #include <xseg/protocol.h>
9 #define MAX_POOL_NAME 64
10 #define MAX_OBJ_NAME 256
12 extern struct log_ctx lc;
22 char pool[MAX_POOL_NAME + 1];
26 char obj_name[MAX_OBJ_NAME + 1];
27 enum rados_state state;
30 void rados_ack_cb(rados_completion_t c, void *arg)
32 struct peer_req *pr = (struct peer_req*) arg;
33 struct peerd *peer = pr->peer;
34 int ret = rados_aio_get_return_value(c);
37 dispatch(peer, pr, pr->req, dispatch_internal);
40 void rados_commit_cb(rados_completion_t c, void *arg)
42 struct peer_req *pr = (struct peer_req*) arg;
43 struct peerd *peer = pr->peer;
44 int ret = rados_aio_get_return_value(c);
47 dispatch(peer, pr, pr->req, dispatch_internal);
50 int do_aio_read(struct peerd *peer, struct peer_req *pr)
52 struct radosd *rados = (struct radosd *) peer->priv;
53 struct xseg_request *req = pr->req;
54 struct rados_io *rio = (struct rados_io *) pr->priv;
55 char *data = xseg_get_data(peer->xseg, pr->req);
58 rados_completion_t rados_compl;
59 r = rados_aio_create_completion(pr, rados_ack_cb, NULL, &rados_compl);
62 r = rados_aio_read(rados->ioctx, rio->obj_name, rados_compl,
64 req->size - req->serviced,
65 req->offset + req->serviced);
67 rados_aio_release(rados_compl);
73 int do_aio_write(struct peerd *peer, struct peer_req *pr)
75 struct radosd *rados = (struct radosd *) peer->priv;
76 struct xseg_request *req = pr->req;
77 struct rados_io *rio = (struct rados_io *) pr->priv;
78 char *data = xseg_get_data(peer->xseg, pr->req);
81 rados_completion_t rados_compl;
82 r = rados_aio_create_completion(pr, NULL, rados_commit_cb, &rados_compl);
85 r = rados_aio_write(rados->ioctx, rio->obj_name, rados_compl,
87 req->size - req->serviced,
88 req->offset + req->serviced);
90 rados_aio_release(rados_compl);
96 int handle_delete(struct peerd *peer, struct peer_req *pr)
99 struct radosd *rados = (struct radosd *) peer->priv;
100 struct rados_io *rio = (struct rados_io *) pr->priv;
102 //log_pr("delete start", pr);
103 XSEGLOG2(&lc, I, "Deleting %s", rio->obj_name);
104 r = rados_remove(rados->ioctx, rio->obj_name);
107 XSEGLOG2(&lc, E, "Deletion of %s failed", rio->obj_name);
112 XSEGLOG2(&lc, E, "Deletion of %s completed", rio->obj_name);
118 int handle_info(struct peerd *peer, struct peer_req *pr)
123 struct xseg_request *req = pr->req;
124 struct radosd *rados = (struct radosd *) peer->priv;
125 struct rados_io *rio = (struct rados_io *) pr->priv;
126 char *req_data = xseg_get_data(peer->xseg, req);
127 struct xseg_reply_info *xinfo = req_data;
129 XSEGLOG2(&lc, I, "Getting info of %s", rio->obj_name);
130 r = rados_stat(rados->ioctx, rio->obj_name, &size, &pmtime);
133 XSEGLOG2(&lc, I, "Getting info of %s failed", rio->obj_name);
138 pr->retval = sizeof(uint64_t);
139 XSEGLOG2(&lc, I, "Getting info of %s completed", rio->obj_name);
145 //FIXME req->state no longer apply
146 int handle_read(struct peerd *peer, struct peer_req *pr)
148 struct rados_io *rio = (struct rados_io *) (pr->priv);
149 struct xseg_request *req = pr->req;
151 if (rio->state == ACCEPTED) {
156 //should we ensure req->op = X_READ ?
157 rio->state = PENDING;
158 XSEGLOG2(&lc, I, "Reading %s", rio->obj_name);
159 if (do_aio_read(peer, pr) < 0) {
160 XSEGLOG2(&lc, I, "Reading of %s failed on do_aio_read", rio->obj_name);
164 else if (rio->state == PENDING) {
165 XSEGLOG2(&lc, I, "Reading of %s callback", rio->obj_name);
166 data = xseg_get_data(peer->xseg, pr->req);
168 req->serviced += pr->retval;
169 else if (pr->retval == 0) {
170 XSEGLOG2(&lc, I, "Reading of %s reached end of file at %llu bytes. Zeroing out rest",
171 rio->obj_name, (unsigned long long) req->serviced);
172 /* reached end of object. zero out rest of data
173 * requested from this object
175 memset(data + req->serviced, 0, req->datalen - req->serviced);
176 req->serviced = req->datalen ;
178 else if (pr->retval == -2) {
179 XSEGLOG2(&lc, I, "Reading of %s return -2. Zeroing out data", rio->obj_name);
180 /* object not found. return zeros instead */
181 memset(data, 0, req->datalen);
182 req->serviced = req->datalen;
185 XSEGLOG2(&lc, E, "Reading of %s failed", rio->obj_name);
186 /* pr->retval < 0 && pr->retval != -2 */
190 if (req->serviced >= req->datalen) {
191 XSEGLOG2(&lc, I, "Reading of %s completed", rio->obj_name);
197 /* should not happen */
201 //TODO assert req->op == X_READ
204 XSEGLOG2(&lc, I, "Resubmitting read of %s", rio->obj_name);
205 if (do_aio_read(peer, pr) < 0) {
206 XSEGLOG2(&lc, E, "Reading of %s failed on do_aio_read", rio->obj_name);
211 /* should not reach this */
212 printf("read request reached this\n");
218 int handle_write(struct peerd *peer, struct peer_req *pr)
220 struct rados_io *rio = (struct rados_io *) (pr->priv);
221 struct xseg_request *req = pr->req;
222 if (rio->state == ACCEPTED) {
225 if (req->flags & XF_FLUSH) {
234 //should we ensure req->op = X_READ ?
235 rio->state = PENDING;
236 XSEGLOG2(&lc, I, "Writing %s", rio->obj_name);
237 if (do_aio_write(peer, pr) < 0) {
238 XSEGLOG2(&lc, E, "Writing of %s failed on do_aio_write", rio->obj_name);
242 else if (rio->state == PENDING) {
243 /* rados writes return 0 if write succeeded or < 0 if failed
244 * no resubmission occurs
246 XSEGLOG2(&lc, I, "Writing of %s callback", rio->obj_name);
247 if (pr->retval == 0) {
248 XSEGLOG2(&lc, I, "Writing of %s completed", rio->obj_name);
249 req->serviced = req->datalen;
254 XSEGLOG2(&lc, I, "Writing of %s failed", rio->obj_name);
260 /* should not reach this */
261 printf("write request reached this\n");
267 int handle_copy(struct peerd *peer, struct peer_req *pr)
269 struct radosd *rados = (struct radosd *) peer->priv;
270 struct xseg_request *req = pr->req;
271 struct rados_io *rio = (struct rados_io *) pr->priv;
273 char *buf, src_name[MAX_OBJ_NAME + 1];
274 struct xseg_request_copy *xcopy = xseg_get_data(peer->xseg, req);
275 unsigned int end = (xcopy->targetlen > MAX_OBJ_NAME) ? MAX_OBJ_NAME : xcopy->targetlen;
277 strncpy(src_name, xcopy->target, end);
281 buf = malloc(req->size);
286 XSEGLOG2(&lc, I, "Copy of object %s to object %s started", src_name, rio->obj_name);
289 r = rados_read(rados->ioctx, src_name, buf, req->size, 0);
291 XSEGLOG2(&lc, E, "Read of object %s failed", src_name);
295 memset(buf+r, 0, req->size - r);
299 } while (sum < req->size);
300 XSEGLOG2(&lc, D, "Read of object %s Completed", src_name);
302 r = rados_write_full(rados->ioctx, rio->obj_name, buf, req->size);
304 XSEGLOG2(&lc, E, "Read of object %s failed", rio->obj_name);
309 req->serviced = req->size;
310 XSEGLOG2(&lc, I, "Copy of object %s to object %s completed", src_name, rio->obj_name);
317 XSEGLOG2(&lc, E, "Copy of object %s to object %s failed", src_name, rio->obj_name);
322 int handle_open(struct peerd *peer, struct peer_req *pr)
324 /* FIXME to be implemented */
330 int handle_close(struct peerd *peer, struct peer_req *pr)
332 /* FIXME to be implemented */
337 int custom_peer_init(struct peerd *peer, int argc, char *argv[])
340 struct radosd *rados = malloc(sizeof(struct radosd));
341 struct rados_io *rio;
347 for (i = 0; i < argc; i++) {
348 if (!strcmp(argv[i], "--pool") && (i+1) < argc){
349 strncpy(rados->pool, argv[i+1], MAX_POOL_NAME);
350 rados->pool[MAX_POOL_NAME] = 0;
355 if (!rados->pool[0]){
360 if (rados_create(&rados->cluster, NULL) < 0) {
361 printf("Rados create failed!\n");
364 if (rados_conf_read_file(rados->cluster, NULL) < 0){
365 printf("Error reading rados conf files!\n");
368 if (rados_connect(rados->cluster) < 0) {
369 printf("Rados connect failed!\n");
370 rados_shutdown(rados->cluster);
374 if (rados_pool_lookup(rados->cluster, rados->pool) < 0) {
375 printf( "Pool does not exists. I will try to create it\n");
376 if (rados_pool_create(rados->cluster, rados->pool) < 0){
377 printf("Couldn't create pool!\n");
378 rados_shutdown(rados->cluster);
382 printf("Pool created.\n");
384 if (rados_ioctx_create(rados->cluster, rados->pool, &(rados->ioctx)) < 0) {
385 printf("ioctx create problem.\n");
386 rados_shutdown(rados->cluster);
390 peer->priv = (void *) rados;
391 for (i = 0; i < peer->nr_ops; i++) {
392 rio = malloc(sizeof(struct rados_io));
395 for (j = 0; j < i; j++) {
396 free(peer->peer_reqs[j].priv);
402 peer->peer_reqs[i].priv = (void *) rio;
407 // nothing to do here for now
408 int custom_arg_parse(int argc, const char *argv[])
413 int dispatch(struct peerd *peer, struct peer_req *pr, struct xseg_request *req,
414 enum dispatch_reason reason)
416 struct rados_io *rio = (struct rados_io *) (pr->priv);
417 char *target = xseg_get_target(peer->xseg, pr->req);
418 unsigned int end = (pr->req->targetlen > MAX_OBJ_NAME) ? MAX_OBJ_NAME : pr->req->targetlen;
419 strncpy(rio->obj_name, target, end);
420 rio->obj_name[end] = 0;
421 //log_pr("dispatch", pr);
422 if (reason == dispatch_accept)
423 rio->state = ACCEPTED;
425 switch (pr->req->op){
427 handle_read(peer, pr); break;
429 handle_write(peer, pr); break;
432 defer_request(peer, pr);
434 handle_delete(peer, pr);
438 defer_request(peer, pr);
440 handle_info(peer, pr);
444 defer_request(peer, pr);
446 handle_copy(peer, pr);
449 handle_open(peer, pr); break;
451 handle_close(peer, pr); break;