6 #include <rados/librados.h>
7 #include <xseg/protocol.h>
9 #define MAX_POOL_NAME 64
10 #define MAX_OBJ_NAME 256
15 char pool[MAX_POOL_NAME];
19 char obj_name[MAX_OBJ_NAME];
22 void rados_ack_cb(rados_completion_t c, void *arg)
24 struct peer_req *pr = (struct peer_req*) arg;
25 struct peerd *peer = pr->peer;
26 int ret = rados_aio_get_return_value(c);
29 dispatch(peer, pr, pr->req);
32 void rados_commit_cb(rados_completion_t c, void *arg)
34 struct peer_req *pr = (struct peer_req*) arg;
35 struct peerd *peer = pr->peer;
36 int ret = rados_aio_get_return_value(c);
39 dispatch(peer, pr, pr->req);
42 int do_aio_read(struct peerd *peer, struct peer_req *pr)
44 struct radosd *rados = (struct radosd *) peer->priv;
45 struct xseg_request *req = pr->req;
46 struct rados_io *rio = (struct rados_io *) pr->priv;
47 char *data = xseg_get_data(peer->xseg, pr->req);
50 rados_completion_t rados_compl;
51 r = rados_aio_create_completion(pr, rados_ack_cb, NULL, &rados_compl);
54 r = rados_aio_read(rados->ioctx, rio->obj_name, rados_compl,
56 req->size - req->serviced,
57 req->offset + req->serviced);
59 rados_aio_release(rados_compl);
65 int do_aio_write(struct peerd *peer, struct peer_req *pr)
67 struct radosd *rados = (struct radosd *) peer->priv;
68 struct xseg_request *req = pr->req;
69 struct rados_io *rio = (struct rados_io *) pr->priv;
70 char *data = xseg_get_data(peer->xseg, pr->req);
73 rados_completion_t rados_compl;
74 r = rados_aio_create_completion(pr, NULL, rados_commit_cb, &rados_compl);
77 r = rados_aio_write(rados->ioctx, rio->obj_name, rados_compl,
79 req->size - req->serviced,
80 req->offset + req->serviced);
82 rados_aio_release(rados_compl);
88 int handle_delete(struct peerd *peer, struct peer_req *pr)
91 struct radosd *rados = (struct radosd *) peer->priv;
92 struct rados_io *rio = (struct rados_io *) pr->priv;
94 //log_pr("delete start", pr);
95 r = rados_remove(rados->ioctx, rio->obj_name);
107 int handle_info(struct peerd *peer, struct peer_req *pr)
112 struct xseg_request *req = pr->req;
113 struct radosd *rados = (struct radosd *) peer->priv;
114 struct rados_io *rio = (struct rados_io *) pr->priv;
115 char *req_data = xseg_get_data(peer->xseg, req);
116 struct xseg_reply_info *xinfo = req_data;
118 log_pr("info start", pr);
120 r = rados_stat(rados->ioctx, rio->obj_name, &size, &pmtime);
127 pr->retval = sizeof(uint64_t);
133 int handle_read(struct peerd *peer, struct peer_req *pr)
135 struct xseg_request *req = pr->req;
137 if (req->state == XS_ACCEPTED) {
142 //should we ensure req->op = X_READ ?
145 if (do_aio_read(peer, pr) < 0) {
149 else if (req->state == XS_PENDING) {
150 data = xseg_get_data(peer->xseg, pr->req);
152 req->serviced += pr->retval;
153 else if (pr->retval == 0) {
154 /* reached end of object. zero out rest of data
155 * requested from this object
157 memset(data, 0, req->datalen - req->serviced);
158 req->serviced = req->datalen ;
160 else if (pr->retval == -2) {
161 /* object not found. return zeros instead */
162 memset(data, 0, req->datalen);
163 req->serviced = req->datalen;
166 /* pr->retval < 0 && pr->retval != -2 */
170 if (req->serviced >= req->datalen) {
171 log_pr("read complete", pr);
177 /* should not happen */
181 //TODO assert req->op == X_READ
184 log_pr("read resubmit", pr);
185 if (do_aio_read(peer, pr) < 0) {
190 /* should not reach this */
191 printf("read request reached this\n");
197 int handle_write(struct peerd *peer, struct peer_req *pr)
199 struct xseg_request *req = pr->req;
200 if (req->state == XS_ACCEPTED) {
203 if (req->flags & XF_FLUSH) {
212 //should we ensure req->op = X_READ ?
214 //log_pr("write", pr);
215 if (do_aio_write(peer, pr) < 0) {
219 else if (req->state == XS_PENDING) {
220 /* rados writes return 0 if write succeeded or < 0 if failed
221 * no resubmission occurs
223 //log_pr("write complete", pr);
224 if (pr->retval == 0) {
225 req->serviced = req->datalen;
235 /* should not reach this */
236 printf("write request reached this\n");
242 int handle_copy(struct peerd *peer, struct peer_req *pr)
248 int custom_peer_init(struct peerd *peer, int argc, char *argv[])
251 struct radosd *rados = malloc(sizeof(struct radosd));
252 struct rados_io *rio;
257 //TODO this should be a parameter. maybe -r (from rados)?
258 strncpy(rados->pool, "xseg", MAX_POOL_NAME);
259 if (rados_create(&rados->cluster, NULL) < 0) {
260 printf("Rados create failed!\n");
263 if (rados_conf_read_file(rados->cluster, NULL) < 0){
264 printf("Error reading rados conf files!\n");
267 if (rados_connect(rados->cluster) < 0) {
268 printf("Rados connect failed!\n");
269 rados_shutdown(rados->cluster);
273 if (rados_pool_lookup(rados->cluster, rados->pool) < 0) {
274 printf( "Pool does not exists. I will try to create it\n");
275 if (rados_pool_create(rados->cluster, rados->pool) < 0){
276 printf("Couldn't create pool!\n");
277 rados_shutdown(rados->cluster);
281 printf("Pool created.\n");
283 if (rados_ioctx_create(rados->cluster, rados->pool, &(rados->ioctx)) < 0) {
284 printf("ioctx create problem.\n");
285 rados_shutdown(rados->cluster);
289 peer->priv = (void *) rados;
290 for (i = 0; i < peer->nr_ops; i++) {
291 rio = malloc(sizeof(struct rados_io));
294 for (j = 0; j < i; j++) {
295 free(peer->peer_reqs[j].priv);
301 peer->peer_reqs[i].priv = (void *) rio;
306 // nothing to do here for now
307 int custom_arg_parse(int argc, const char *argv[])
312 int dispatch(struct peerd *peer, struct peer_req *pr, struct xseg_request *req)
314 struct rados_io *rio = (struct rados_io *) (pr->priv);
315 char *target = xseg_get_target(peer->xseg, pr->req);
316 unsigned int end = (pr->req->targetlen > MAX_OBJ_NAME -1 )? MAX_OBJ_NAME - 1 : pr->req->targetlen;
317 strncpy(rio->obj_name, target, end);
318 rio->obj_name[end] = 0;
319 //log_pr("dispatch", pr);
320 switch (pr->req->op){
322 handle_read(peer, pr); break;
324 handle_write(peer, pr); break;
327 defer_request(peer, pr);
329 handle_delete(peer, pr);
333 defer_request(peer, pr);
335 handle_info(peer, pr);
339 defer_request(peer, pr);
341 handle_copy(peer, pr);