← Back to branch summary

~ubuntu-branches/ubuntu/quantal/ceph/quantal

« back to all changes in this revision

Viewing changes to src/librados.cc

  • Committer: Bazaar Package Importer
  • Author(s): Clint Byrum, Clint Byrum, Micah Gersten
  • Date: 2011-02-12 22:50:26 UTC
  • mfrom: (1.1.1 upstream)
  • Revision ID: james.westby@ubuntu.com-20110212225026-yyyw4tk0msgql3ul
Tags: 0.24.2-0ubuntu1
https://launchpad.net/bugs/658670
https://launchpad.net/bugs/684011
[ Clint Byrum <clint@ubuntu.com> ]
* New upstream release. (LP: #658670, LP: #684011)
* debian/patches/fix-mkcephfs.patch: dropped (applied upstream)
* Removed .la files from libceph1-dev, libcrush1-dev and 
  librados1-dev (per Debian policy v3.9.1 10.2).
* debian/control: adding pkg-config as a build dependency
* debian/control: depend on libcrypto++-dev instead of libssl-dev
* debian/watch: added watch file

[ Micah Gersten <micahg@ubuntu.com> ]
* debian/control: add Homepage

Show diffs side-by-side

added added

removed removed

Lines of Context:
src/librados.cc
72
72
 
73
73
  Mutex lock;
74
74
  Cond cond;
 
75
  SafeTimer timer;
75
76
 
76
77
 
77
78
public:
78
 
  RadosClient() : messenger(NULL), lock("radosclient") {
 
79
  RadosClient() : messenger(NULL), lock("radosclient"), timer(lock) {
79
80
    messenger = new SimpleMessenger();
80
81
  }
81
82
 
82
83
  ~RadosClient();
83
 
  bool init();
 
84
  int init();
84
85
  void shutdown();
85
86
 
86
87
  struct PoolCtx {
134
135
  int write(PoolCtx& pool, const object_t& oid, off_t off, bufferlist& bl, size_t len);
135
136
  int write_full(PoolCtx& pool, const object_t& oid, bufferlist& bl);
136
137
  int read(PoolCtx& pool, const object_t& oid, off_t off, bufferlist& bl, size_t len);
 
138
  int mapext(PoolCtx& pool, const object_t& oid, off_t off, size_t len, std::map<off_t,size_t>& m);
 
139
  int sparse_read(PoolCtx& pool, const object_t& oid, off_t off, size_t len, std::map<off_t,size_t>& m, bufferlist& bl);
137
140
  int remove(PoolCtx& pool, const object_t& oid);
138
141
  int stat(PoolCtx& pool, const object_t& oid, uint64_t *psize, time_t *pmtime);
139
142
  int trunc(PoolCtx& pool, const object_t& oid, size_t size);
144
147
  int getxattr(PoolCtx& pool, const object_t& oid, const char *name, bufferlist& bl);
145
148
  int setxattr(PoolCtx& pool, const object_t& oid, const char *name, bufferlist& bl);
146
149
  int getxattrs(PoolCtx& pool, const object_t& oid, map<string, bufferlist>& attrset);
 
150
  int rmxattr(PoolCtx& pool, const object_t& oid, const char *name);
147
151
 
148
152
  int list_pools(std::list<string>& ls);
149
153
  int get_pool_stats(std::list<string>& ls, map<string,pool_stat_t>& result);
330
334
  }
331
335
};
332
336
 
333
 
bool RadosClient::init()
 
337
int RadosClient::init()
334
338
{
335
339
  // get monmap
336
 
  if (monclient.build_initial_monmap() < 0)
337
 
    return false;
 
340
  int ret = monclient.build_initial_monmap();
 
341
  if (ret < 0)
 
342
    return ret;
 
343
 
 
344
  assert_warn(messenger);
 
345
  if (!messenger)
 
346
    return -ENOMEM;
338
347
 
339
348
  dout(1) << "starting msgr at " << messenger->get_ms_addr() << dendl;
340
349
 
341
350
  messenger->register_entity(entity_name_t::CLIENT(-1));
342
 
  assert_warn(messenger);
343
 
  if (!messenger)
344
 
    return false;
345
351
  dout(1) << "starting objecter" << dendl;
346
352
 
347
 
  objecter = new Objecter(messenger, &monclient, &osdmap, lock);
 
353
  objecter = new Objecter(messenger, &monclient, &osdmap, lock, timer);
348
354
  if (!objecter)
349
 
    return false;
 
355
    return -ENOMEM;
 
356
  objecter->set_balanced_budget();
350
357
 
351
358
  monclient.set_messenger(messenger);
352
359
  
357
364
 
358
365
  dout(1) << "setting wanted keys" << dendl;
359
366
  monclient.set_want_keys(CEPH_ENTITY_TYPE_MON | CEPH_ENTITY_TYPE_OSD);
360
 
  dout(1) << "iit" << dendl;
 
367
  dout(1) << "calling monclient init" << dendl;
361
368
  monclient.init();
362
369
 
363
370
  int err = monclient.authenticate(g_conf.client_mount_timeout);
364
371
  if (err) {
365
372
    dout(0) << *g_conf.entity_name << " authentication error " << strerror(-err) << dendl;
366
 
    return false;
 
373
    return err;
367
374
  }
368
375
  messenger->set_myname(entity_name_t::CLIENT(monclient.get_global_id()));
369
376
 
370
377
  lock.Lock();
371
378
 
 
379
  timer.init();
 
380
 
372
381
  objecter->set_client_incarnation(0);
373
382
  objecter->init();
374
383
  monclient.renew_subs();
381
390
 
382
391
  dout(1) << "init done" << dendl;
383
392
 
384
 
  return true;
 
393
  return 0;
385
394
}
386
395
 
387
396
void RadosClient::shutdown()
388
397
{
389
398
  lock.Lock();
 
399
  monclient.shutdown();
390
400
  objecter->shutdown();
 
401
  timer.shutdown();
391
402
  lock.Unlock();
392
403
  messenger->shutdown();
393
404
  messenger->wait();
578
589
{
579
590
  int reply;
580
591
  PoolCtx* ctx = (PoolCtx *) pool;
581
 
  ceph_object_layout layout = objecter->osdmap
582
 
    ->make_object_layout(oid, ctx->poolid);
 
592
 
 
593
  object_locator_t oloc(ctx->poolid);
583
594
 
584
595
  Mutex mylock("RadosClient::snap_rollback::mylock");
585
596
  Cond cond;
587
598
  Context *onack = new C_SafeCond(&mylock, &cond, &done, &reply);
588
599
 
589
600
  lock.Lock();
590
 
  objecter->rollback_object(oid, layout, snapc, snapid,
 
601
  objecter->rollback_object(oid, oloc, snapc, snapid,
591
602
                     g_clock.now(), onack, NULL);
592
603
  lock.Unlock();
593
604
 
799
810
  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
800
811
 
801
812
  lock.Lock();
802
 
  ceph_object_layout layout = objecter->osdmap->make_object_layout(oid, pool.poolid);
803
 
  objecter->create(oid, layout,
 
813
  object_locator_t oloc(pool.poolid);
 
814
  objecter->create(oid, oloc,
804
815
                  pool.snapc, ut, 0, (exclusive ? CEPH_OSD_OP_FLAG_EXCL : 0),
805
816
                  onack, NULL);
806
817
  lock.Unlock();
852
863
  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
853
864
 
854
865
  lock.Lock();
855
 
  ceph_object_layout layout = objecter->osdmap->make_object_layout(oid, pool.poolid);
856
 
  objecter->write(oid, layout,
 
866
  object_locator_t oloc(pool.poolid);
 
867
  objecter->write(oid, oloc,
857
868
                  off, len, pool.snapc, bl, ut, 0,
858
869
                  onack, NULL);
859
870
  lock.Unlock();
885
896
  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
886
897
 
887
898
  lock.Lock();
888
 
  ceph_object_layout layout = objecter->osdmap->make_object_layout(oid, pool.poolid);
889
 
  objecter->write_full(oid, layout,
 
899
  object_locator_t oloc(pool.poolid);
 
900
  objecter->write_full(oid, oloc,
890
901
                  pool.snapc, bl, ut, 0,
891
902
                  onack, NULL);
892
903
  lock.Unlock();
908
919
  c->pbl = pbl;
909
920
 
910
921
  Mutex::Locker l(lock);
911
 
  ceph_object_layout layout = objecter->osdmap->make_object_layout(oid, pool.poolid);
912
 
  objecter->read(oid, layout,
 
922
  object_locator_t oloc(pool.poolid);
 
923
  objecter->read(oid, oloc,
913
924
                 off, len, pool.snap_seq, &c->bl, 0,
914
925
                  onack);
915
926
 
924
935
  c->maxlen = len;
925
936
 
926
937
  Mutex::Locker l(lock);
927
 
  ceph_object_layout layout = objecter->osdmap->make_object_layout(oid, pool.poolid);
928
 
  objecter->read(oid, layout,
 
938
  object_locator_t oloc(pool.poolid);
 
939
  objecter->read(oid, oloc,
929
940
                 off, len, pool.snap_seq, &c->bl, 0,
930
941
                  onack);
931
942
 
942
953
  Context *onsafe = new C_aio_Safe(c);
943
954
 
944
955
  Mutex::Locker l(lock);
945
 
  ceph_object_layout layout = objecter->osdmap->make_object_layout(oid, pool.poolid);
946
 
  objecter->write(oid, layout,
 
956
  object_locator_t oloc(pool.poolid);
 
957
  objecter->write(oid, oloc,
947
958
                  off, len, pool.snapc, bl, ut, 0,
948
959
                  onack, onsafe);
949
960
 
960
971
  Context *onsafe = new C_aio_Safe(c);
961
972
 
962
973
  Mutex::Locker l(lock);
963
 
  ceph_object_layout layout = objecter->osdmap->make_object_layout(oid, pool.poolid);
964
 
  objecter->write_full(oid, layout,
 
974
  object_locator_t oloc(pool.poolid);
 
975
  objecter->write_full(oid, oloc,
965
976
                  pool.snapc, bl, ut, 0,
966
977
                  onack, onsafe);
967
978
 
980
991
  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
981
992
 
982
993
  lock.Lock();
983
 
  ceph_object_layout layout = objecter->osdmap->make_object_layout(oid, pool.poolid);
984
 
  objecter->remove(oid, layout,
 
994
  object_locator_t oloc(pool.poolid);
 
995
  objecter->remove(oid, oloc,
985
996
                  snapc, ut, 0,
986
997
                  onack, NULL);
987
998
  lock.Unlock();
1010
1021
  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
1011
1022
 
1012
1023
  lock.Lock();
1013
 
  ceph_object_layout layout = objecter->osdmap->make_object_layout(oid, pool.poolid);
1014
 
  objecter->trunc(oid, layout,
 
1024
  object_locator_t oloc(pool.poolid);
 
1025
  objecter->trunc(oid, oloc,
1015
1026
                  pool.snapc, ut, 0,
1016
1027
                  size, 0,
1017
1028
                  onack, NULL);
1039
1050
 
1040
1051
  lock.Lock();
1041
1052
  SnapContext snapc;
1042
 
  ceph_object_layout layout = objecter->osdmap->make_object_layout(oid, pool.poolid);
 
1053
  object_locator_t oloc(pool.poolid);
1043
1054
  ObjectOperation wr;
1044
1055
  wr.tmap_update(cmdbl);
1045
 
  objecter->mutate(oid, layout, wr, snapc, ut, 0, onack, NULL);
 
1056
  objecter->mutate(oid, oloc, wr, snapc, ut, 0, onack, NULL);
1046
1057
  lock.Unlock();
1047
1058
 
1048
1059
  mylock.Lock();
1067
1078
 
1068
1079
 
1069
1080
  lock.Lock();
1070
 
  ceph_object_layout layout = objecter->osdmap->make_object_layout(oid, pool.poolid);
 
1081
  object_locator_t oloc(pool.poolid);
1071
1082
  ObjectOperation rd;
1072
1083
  rd.call(cls, method, inbl);
1073
 
  objecter->read(oid, layout, rd, pool.snap_seq, &outbl, 0, onack);
 
1084
  objecter->read(oid, oloc, rd, pool.snap_seq, &outbl, 0, onack);
1074
1085
  lock.Unlock();
1075
1086
 
1076
1087
  mylock.Lock();
1092
1103
  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
1093
1104
 
1094
1105
  lock.Lock();
1095
 
  ceph_object_layout layout = objecter->osdmap->make_object_layout(oid, pool.poolid);
1096
 
  objecter->read(oid, layout,
 
1106
  object_locator_t oloc(pool.poolid);
 
1107
  objecter->read(oid, oloc,
1097
1108
              off, len, pool.snap_seq, &bl, 0,
1098
1109
              onack);
1099
1110
  lock.Unlock();
1115
1126
  return bl.length();
1116
1127
}
1117
1128
 
 
1129
int RadosClient::mapext(PoolCtx& pool, const object_t& oid, off_t off, size_t len, std::map<off_t,size_t>& m)
 
1130
{
 
1131
  SnapContext snapc;
 
1132
  bufferlist bl;
 
1133
 
 
1134
  Mutex mylock("RadosClient::read::mylock");
 
1135
  Cond cond;
 
1136
  bool done;
 
1137
  int r;
 
1138
  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
 
1139
 
 
1140
  lock.Lock();
 
1141
  object_locator_t oloc(pool.poolid);
 
1142
  objecter->mapext(oid, oloc,
 
1143
              off, len, pool.snap_seq, &bl, 0,
 
1144
              onack);
 
1145
  lock.Unlock();
 
1146
 
 
1147
  mylock.Lock();
 
1148
  while (!done)
 
1149
    cond.Wait(mylock);
 
1150
  mylock.Unlock();
 
1151
  dout(10) << "Objecter returned from read r=" << r << dendl;
 
1152
 
 
1153
  if (r < 0)
 
1154
    return r;
 
1155
 
 
1156
  bufferlist::iterator iter = bl.begin();
 
1157
  ::decode(m, iter);
 
1158
 
 
1159
  return m.size();
 
1160
}
 
1161
 
 
1162
int RadosClient::sparse_read(PoolCtx& pool, const object_t& oid, off_t off, size_t len,
 
1163
                             std::map<off_t,size_t>& m, bufferlist& data_bl)
 
1164
{
 
1165
  SnapContext snapc;
 
1166
  bufferlist bl;
 
1167
 
 
1168
  Mutex mylock("RadosClient::read::mylock");
 
1169
  Cond cond;
 
1170
  bool done;
 
1171
  int r;
 
1172
  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
 
1173
 
 
1174
  lock.Lock();
 
1175
  object_locator_t oloc(pool.poolid);
 
1176
  objecter->sparse_read(oid, oloc,
 
1177
              off, len, pool.snap_seq, &bl, 0,
 
1178
              onack);
 
1179
  lock.Unlock();
 
1180
 
 
1181
  mylock.Lock();
 
1182
  while (!done)
 
1183
    cond.Wait(mylock);
 
1184
  mylock.Unlock();
 
1185
  dout(10) << "Objecter returned from read r=" << r << dendl;
 
1186
 
 
1187
  if (r < 0)
 
1188
    return r;
 
1189
 
 
1190
  bufferlist::iterator iter = bl.begin();
 
1191
  ::decode(m, iter);
 
1192
  ::decode(data_bl, iter);
 
1193
 
 
1194
  return m.size();
 
1195
}
 
1196
 
1118
1197
int RadosClient::stat(PoolCtx& pool, const object_t& oid, uint64_t *psize, time_t *pmtime)
1119
1198
{
1120
1199
  SnapContext snapc;
1131
1210
    psize = &size;
1132
1211
 
1133
1212
  lock.Lock();
1134
 
  ceph_object_layout layout = objecter->osdmap->make_object_layout(oid, pool.poolid);
1135
 
  objecter->stat(oid, layout,
 
1213
  object_locator_t oloc(pool.poolid);
 
1214
  objecter->stat(oid, oloc,
1136
1215
              pool.snap_seq, psize, &mtime, 0,
1137
1216
              onack);
1138
1217
  lock.Unlock();
1161
1240
  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
1162
1241
 
1163
1242
  lock.Lock();
1164
 
  ceph_object_layout layout = objecter->osdmap->make_object_layout(oid, pool.poolid);
1165
 
  objecter->getxattr(oid, layout,
 
1243
  object_locator_t oloc(pool.poolid);
 
1244
  objecter->getxattr(oid, oloc,
1166
1245
              name, pool.snap_seq, &bl, 0,
1167
1246
              onack);
1168
1247
  lock.Unlock();
1179
1258
  return bl.length();
1180
1259
}
1181
1260
 
 
1261
int RadosClient::rmxattr(PoolCtx& pool, const object_t& oid, const char *name)
 
1262
{
 
1263
  utime_t ut = g_clock.now();
 
1264
 
 
1265
  /* can't write to a snapshot */
 
1266
  if (pool.snap_seq != CEPH_NOSNAP)
 
1267
    return -EINVAL;
 
1268
 
 
1269
  Mutex mylock("RadosClient::rmxattr::mylock");
 
1270
  Cond cond;
 
1271
  bool done;
 
1272
  int r;
 
1273
 
 
1274
  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
 
1275
 
 
1276
  object_locator_t oloc(pool.poolid);
 
1277
 
 
1278
  lock.Lock();
 
1279
  objecter->removexattr(oid, oloc, name,
 
1280
                  pool.snapc, ut, 0,
 
1281
                  onack, NULL);
 
1282
  lock.Unlock();
 
1283
 
 
1284
  mylock.Lock();
 
1285
  while (!done)
 
1286
    cond.Wait(mylock);
 
1287
  mylock.Unlock();
 
1288
 
 
1289
  if (r < 0)
 
1290
    return r;
 
1291
 
 
1292
  return 0;
 
1293
}
 
1294
 
1182
1295
int RadosClient::setxattr(PoolCtx& pool, const object_t& oid, const char *name, bufferlist& bl)
1183
1296
{
1184
1297
  utime_t ut = g_clock.now();
1195
1308
  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
1196
1309
 
1197
1310
  lock.Lock();
1198
 
  ceph_object_layout layout = objecter->osdmap->make_object_layout(oid, pool.poolid);
1199
 
  objecter->setxattr(oid, layout, name,
 
1311
  object_locator_t oloc(pool.poolid);
 
1312
  objecter->setxattr(oid, oloc, name,
1200
1313
                  pool.snapc, bl, ut, 0,
1201
1314
                  onack, NULL);
1202
1315
  lock.Unlock();
1228
1341
  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
1229
1342
 
1230
1343
  lock.Lock();
1231
 
  ceph_object_layout layout = objecter->osdmap->make_object_layout(oid, pool.poolid);
 
1344
  object_locator_t oloc(pool.poolid);
1232
1345
  map<string, bufferlist> aset;
1233
 
  objecter->getxattrs(oid, layout, pool.snap_seq,
 
1346
  objecter->getxattrs(oid, oloc, pool.snap_seq,
1234
1347
                      aset,
1235
1348
                      0, onack);
1236
1349
  lock.Unlock();
1261
1374
Rados::~Rados()
1262
1375
{
1263
1376
  if (client)
1264
 
    delete (RadosClient *)client;
 
1377
    delete client;
1265
1378
  client = NULL;
1266
1379
}
1267
1380
 
1279
1392
  if (g_conf.clock_tare) g_clock.tare();
1280
1393
 
1281
1394
  client = new RadosClient();
1282
 
  return ((RadosClient *)client)->init() ? 0 : -1;
 
1395
  return client->init();
1283
1396
}
1284
1397
 
1285
1398
void Rados::shutdown()
1286
1399
{
1287
 
  ((RadosClient *)client)->shutdown();
 
1400
  client->shutdown();
1288
1401
}
1289
1402
 
1290
1403
int Rados::list_pools(std::list<string>& v)
1291
1404
{
1292
1405
  if (!client)
1293
1406
    return -EINVAL;
1294
 
  return ((RadosClient *)client)->list_pools(v);
 
1407
  return client->list_pools(v);
1295
1408
}
1296
1409
 
1297
1410
int Rados::get_pool_stats(std::list<string>& v, std::map<string,pool_stat_t>& result)
1299
1412
  if (!client)
1300
1413
    return -EINVAL;
1301
1414
  map<string,::pool_stat_t> rawresult;
1302
 
  int r = ((RadosClient *)client)->get_pool_stats(v, rawresult);
 
1415
  int r = client->get_pool_stats(v, rawresult);
1303
1416
  for (map<string,::pool_stat_t>::iterator p = rawresult.begin();
1304
1417
       p != rawresult.end();
1305
1418
       p++) {
1310
1423
    v.num_object_clones = p->second.num_object_clones;
1311
1424
    v.num_object_copies = p->second.num_object_copies;
1312
1425
    v.num_objects_missing_on_primary = p->second.num_objects_missing_on_primary;
 
1426
    v.num_objects_unfound = p->second.num_objects_unfound;
1313
1427
    v.num_objects_degraded = p->second.num_objects_degraded;
1314
1428
    v.num_rd = p->second.num_rd;
1315
1429
    v.num_rd_kb = p->second.num_rd_kb;
1324
1438
  string str(name);
1325
1439
  if (!client)
1326
1440
    return -EINVAL;
1327
 
  return ((RadosClient *)client)->create_pool(str, auid, crush_rule);
 
1441
  return client->create_pool(str, auid, crush_rule);
1328
1442
}
1329
1443
 
1330
1444
int Rados::delete_pool(const rados_pool_t& pool)
1331
1445
{
1332
1446
  if (!client) return -EINVAL;
1333
 
  return ((RadosClient *)client)->delete_pool(pool);
 
1447
  return client->delete_pool(pool);
1334
1448
}
1335
1449
 
1336
1450
int Rados::change_pool_auid(const rados_pool_t& pool, uint64_t auid)
1337
1451
{
1338
1452
  if (!client) return -EINVAL;
1339
 
  return ((RadosClient *)client)->change_pool_auid(pool, auid);
 
1453
  return client->change_pool_auid(pool, auid);
1340
1454
}
1341
1455
 
1342
1456
int Rados::get_fs_stats(statfs_t& result)
1344
1458
  if (!client)
1345
1459
    return -EINVAL;
1346
1460
  ceph_statfs stats;
1347
 
  int r = ((RadosClient *)client)->get_fs_stats(stats);
 
1461
  int r = client->get_fs_stats(stats);
1348
1462
  result.kb = stats.kb;
1349
1463
  result.kb_used = stats.kb_used;
1350
1464
  result.kb_avail = stats.kb_avail;
1369
1483
 
1370
1484
  Objecter::ListContext *h = (Objecter::ListContext *)ctx.ctx;
1371
1485
  h->list.clear();
1372
 
  int r = ((RadosClient *)client)->list(h, max);
 
1486
  int r = client->list(h, max);
1373
1487
  while (!h->list.empty()) {
1374
1488
    entries.push_back(h->list.front().name.c_str());
1375
1489
    h->list.pop_front();
1389
1503
  if (!client)
1390
1504
    return -EINVAL;
1391
1505
  object_t oid(o);
1392
 
  return ((RadosClient *)client)->create(*(RadosClient::PoolCtx *)pool, oid, exclusive);
 
1506
  return client->create(*(RadosClient::PoolCtx *)pool, oid, exclusive);
1393
1507
}
1394
1508
 
1395
1509
int Rados::write(rados_pool_t pool, const string& o, off_t off, bufferlist& bl, size_t len)
1397
1511
  if (!client)
1398
1512
    return -EINVAL;
1399
1513
  object_t oid(o);
1400
 
  return ((RadosClient *)client)->write(*(RadosClient::PoolCtx *)pool, oid, off, bl, len);
 
1514
  return client->write(*(RadosClient::PoolCtx *)pool, oid, off, bl, len);
1401
1515
}
1402
1516
 
1403
1517
int Rados::write_full(rados_pool_t pool, const string& o, bufferlist& bl)
1405
1519
  if (!client)
1406
1520
    return -EINVAL;
1407
1521
  object_t oid(o);
1408
 
  return ((RadosClient *)client)->write_full(*(RadosClient::PoolCtx *)pool, oid, bl);
 
1522
  return client->write_full(*(RadosClient::PoolCtx *)pool, oid, bl);
1409
1523
}
1410
1524
 
1411
1525
int Rados::trunc(rados_pool_t pool, const string& o, size_t size)
1413
1527
  if (!client)
1414
1528
    return -EINVAL;
1415
1529
  object_t oid(o);
1416
 
  return ((RadosClient *)client)->trunc(*(RadosClient::PoolCtx *)pool, oid, size);
 
1530
  return client->trunc(*(RadosClient::PoolCtx *)pool, oid, size);
1417
1531
}
1418
1532
 
1419
1533
int Rados::remove(rados_pool_t pool, const string& o)
1421
1535
  if (!client)
1422
1536
    return -EINVAL;
1423
1537
  object_t oid(o);
1424
 
  return ((RadosClient *)client)->remove(*(RadosClient::PoolCtx *)pool, oid);
 
1538
  return client->remove(*(RadosClient::PoolCtx *)pool, oid);
1425
1539
}
1426
1540
 
1427
1541
int Rados::read(rados_pool_t pool, const string& o, off_t off, bufferlist& bl, size_t len)
1429
1543
  if (!client)
1430
1544
    return -EINVAL;
1431
1545
  object_t oid(o);
1432
 
  return ((RadosClient *)client)->read(*(RadosClient::PoolCtx *)pool, oid, off, bl, len);
 
1546
  return client->read(*(RadosClient::PoolCtx *)pool, oid, off, bl, len);
 
1547
}
 
1548
 
 
1549
int Rados::mapext(rados_pool_t pool, const string& o, off_t off, size_t len, std::map<off_t, size_t>& m)
 
1550
{
 
1551
  if (!client)
 
1552
    return -EINVAL;
 
1553
  object_t oid(o);
 
1554
  return ((RadosClient *)client)->mapext(*(RadosClient::PoolCtx *)pool, oid, off, len, m);
 
1555
}
 
1556
 
 
1557
int Rados::sparse_read(rados_pool_t pool, const string& o, off_t off, size_t len, std::map<off_t, size_t>& m, bufferlist& bl)
 
1558
{
 
1559
  if (!client)
 
1560
    return -EINVAL;
 
1561
  object_t oid(o);
 
1562
  return ((RadosClient *)client)->sparse_read(*(RadosClient::PoolCtx *)pool, oid, off, len, m, bl);
1433
1563
}
1434
1564
 
1435
1565
int Rados::getxattr(rados_pool_t pool, const string& o, const char *name, bufferlist& bl)
1437
1567
  if (!client)
1438
1568
    return -EINVAL;
1439
1569
  object_t oid(o);
1440
 
  return ((RadosClient *)client)->getxattr(*(RadosClient::PoolCtx *)pool, oid, name, bl);
 
1570
  return client->getxattr(*(RadosClient::PoolCtx *)pool, oid, name, bl);
1441
1571
}
1442
1572
 
1443
1573
int Rados::getxattrs(rados_pool_t pool, const string& o, map<std::string, bufferlist>& attrset)
1445
1575
  if (!client)
1446
1576
    return -EINVAL;
1447
1577
  object_t oid(o);
1448
 
  return ((RadosClient *)client)->getxattrs(*(RadosClient::PoolCtx *)pool, oid, attrset);
 
1578
  return client->getxattrs(*(RadosClient::PoolCtx *)pool, oid, attrset);
1449
1579
}
1450
1580
 
1451
1581
int Rados::setxattr(rados_pool_t pool, const string& o, const char *name, bufferlist& bl)
1453
1583
  if (!client)
1454
1584
    return -EINVAL;
1455
1585
  object_t oid(o);
1456
 
  return ((RadosClient *)client)->setxattr(*(RadosClient::PoolCtx *)pool, oid, name, bl);
 
1586
  return client->setxattr(*(RadosClient::PoolCtx *)pool, oid, name, bl);
 
1587
}
 
1588
 
 
1589
int Rados::rmxattr(rados_pool_t pool, const string& o, const char *name)
 
1590
{
 
1591
  if (!client)
 
1592
    return -EINVAL;
 
1593
  object_t oid(o);
 
1594
  return client->rmxattr(*(RadosClient::PoolCtx *)pool, oid, name);
1457
1595
}
1458
1596
 
1459
1597
int Rados::stat(rados_pool_t pool, const string& o, uint64_t *psize, time_t *pmtime)
1461
1599
  if (!client)
1462
1600
    return -EINVAL;
1463
1601
  object_t oid(o);
1464
 
  return ((RadosClient *)client)->stat(*(RadosClient::PoolCtx *)pool, oid, psize, pmtime);
 
1602
  return client->stat(*(RadosClient::PoolCtx *)pool, oid, psize, pmtime);
1465
1603
}
1466
1604
 
1467
1605
int Rados::tmap_update(rados_pool_t pool, const string& o, bufferlist& cmdbl)
1469
1607
  if (!client)
1470
1608
    return -EINVAL;
1471
1609
  object_t oid(o);
1472
 
  return ((RadosClient *)client)->tmap_update(*(RadosClient::PoolCtx *)pool, oid, cmdbl);
 
1610
  return client->tmap_update(*(RadosClient::PoolCtx *)pool, oid, cmdbl);
1473
1611
}
1474
1612
int Rados::exec(rados_pool_t pool, const string& o, const char *cls, const char *method,
1475
1613
                bufferlist& inbl, bufferlist& outbl)
1477
1615
  if (!client)
1478
1616
    return -EINVAL;
1479
1617
  object_t oid(o);
1480
 
  return ((RadosClient *)client)->exec(*(RadosClient::PoolCtx *)pool, oid, cls, method, inbl, outbl);
 
1618
  return client->exec(*(RadosClient::PoolCtx *)pool, oid, cls, method, inbl, outbl);
1481
1619
}
1482
1620
 
1483
1621
int Rados::lookup_pool(const char *name)
1484
1622
{
1485
 
  return ((RadosClient *)client)->lookup_pool(name);
 
1623
  return client->lookup_pool(name);
1486
1624
}
1487
1625
 
1488
1626
int Rados::open_pool(const char *name, rados_pool_t *pool)
1489
1627
{
1490
 
  int poolid = ((RadosClient *)client)->lookup_pool(name);
 
1628
  int poolid = client->lookup_pool(name);
1491
1629
  if (poolid >= 0) {
1492
1630
    RadosClient::PoolCtx *ctx = new RadosClient::PoolCtx(poolid, name, CEPH_NOSNAP);
1493
1631
    if (!ctx)
1510
1648
 
1511
1649
int Rados::snap_create(const rados_pool_t pool, const char *snapname) {
1512
1650
  if (!client) return -EINVAL;
1513
 
  return ((RadosClient *)client)->snap_create(pool, snapname);
 
1651
  return client->snap_create(pool, snapname);
1514
1652
}
1515
1653
 
1516
1654
int Rados::snap_remove(const rados_pool_t pool, const char *snapname) {
1517
1655
  if (!client) return -EINVAL;
1518
 
  return ((RadosClient *)client)->snap_remove(pool, snapname);
 
1656
  return client->snap_remove(pool, snapname);
1519
1657
}
1520
1658
 
1521
1659
int Rados::snap_rollback_object(const rados_pool_t pool, const std::string& oid,
1522
1660
                                const char *snapname) {
1523
1661
  if (!client) return -EINVAL;
1524
 
  return ((RadosClient *)client)->snap_rollback_object(pool, oid, snapname);
 
1662
  return client->snap_rollback_object(pool, oid, snapname);
1525
1663
}
1526
1664
 
1527
1665
int Rados::selfmanaged_snap_create(const rados_pool_t pool, uint64_t *snapid)
1528
1666
{
1529
1667
  if (!client) return -EINVAL;
1530
 
  return ((RadosClient *)client)->selfmanaged_snap_create(pool, snapid);
 
1668
  return client->selfmanaged_snap_create(pool, snapid);
1531
1669
}
1532
1670
 
1533
1671
int Rados::selfmanaged_snap_remove(const rados_pool_t pool,
1534
1672
                                   uint64_t snapid)
1535
1673
{
1536
1674
  if (!client) return -EINVAL;
1537
 
  return ((RadosClient *)client)->selfmanaged_snap_remove(pool, snapid);
 
1675
  return client->selfmanaged_snap_remove(pool, snapid);
1538
1676
}
1539
1677
 
1540
1678
int Rados::selfmanaged_snap_rollback_object(const rados_pool_t pool,
1549
1687
  for (; iter != snapc.snaps.end(); ++iter) {
1550
1688
    sn.snaps.push_back(*iter);
1551
1689
  }
1552
 
  return ((RadosClient *)client)->selfmanaged_snap_rollback_object(pool, oid, sn, snapid);
 
1690
  return client->selfmanaged_snap_rollback_object(pool, oid, sn, snapid);
1553
1691
}
1554
1692
 
1555
1693
void Rados::set_snap(rados_pool_t pool, snap_t seq)
1577
1715
  if (!client)
1578
1716
    return -EINVAL;
1579
1717
  RadosClient::PoolCtx *ctx = (RadosClient::PoolCtx *)pool;
1580
 
  return ((RadosClient *)client)->snap_list(ctx, snaps);
 
1718
  return client->snap_list(ctx, snaps);
1581
1719
}
1582
1720
 
1583
1721
int Rados::snap_lookup(rados_pool_t pool, const char *name, snap_t *snapid)
1585
1723
  if (!client)
1586
1724
    return -EINVAL;
1587
1725
  RadosClient::PoolCtx *ctx = (RadosClient::PoolCtx *)pool;
1588
 
  return ((RadosClient *)client)->snap_lookup(ctx, name, snapid);
 
1726
  return client->snap_lookup(ctx, name, snapid);
1589
1727
}
1590
1728
 
1591
1729
int Rados::snap_get_name(rados_pool_t pool, snap_t snapid, std::string *s)
1593
1731
  if (!client)
1594
1732
    return -EINVAL;
1595
1733
  RadosClient::PoolCtx *ctx = (RadosClient::PoolCtx *)pool;
1596
 
  return ((RadosClient *)client)->snap_get_name(ctx, snapid, s);
 
1734
  return client->snap_get_name(ctx, snapid, s);
1597
1735
}
1598
1736
int Rados::snap_get_stamp(rados_pool_t pool, snap_t snapid, time_t *t)
1599
1737
{
1600
1738
  if (!client)
1601
1739
    return -EINVAL;
1602
1740
  RadosClient::PoolCtx *ctx = (RadosClient::PoolCtx *)pool;
1603
 
  return ((RadosClient *)client)->snap_get_stamp(ctx, snapid, t);
 
1741
  return client->snap_get_stamp(ctx, snapid, t);
1604
1742
}
1605
1743
 
1606
1744
// AIO
1609
1747
{
1610
1748
  RadosClient::PoolCtx *ctx = (RadosClient::PoolCtx *)pool;
1611
1749
  RadosClient::AioCompletion *pc = (RadosClient::AioCompletion *)c->pc;
1612
 
  int r = ((RadosClient *)client)->aio_read(*ctx, oid, off, pbl, len, pc);
 
1750
  int r = client->aio_read(*ctx, oid, off, pbl, len, pc);
1613
1751
  return r;
1614
1752
}
1615
1753
 
1618
1756
{
1619
1757
  RadosClient::PoolCtx *ctx = (RadosClient::PoolCtx *)pool;
1620
1758
  RadosClient::AioCompletion *pc = (RadosClient::AioCompletion *)c->pc;
1621
 
  int r = ((RadosClient *)client)->aio_write(*ctx, oid, off, bl, len, pc);
 
1759
  int r = client->aio_write(*ctx, oid, off, bl, len, pc);
1622
1760
  return r;
1623
1761
}
1624
1762
 
1625
1763
Rados::AioCompletion *Rados::aio_create_completion()
1626
1764
{
1627
 
  RadosClient::AioCompletion *c = ((RadosClient *)client)->aio_create_completion();
 
1765
  RadosClient::AioCompletion *c = client->aio_create_completion();
1628
1766
  return new AioCompletion(c);
1629
1767
}
1630
1768
 
1631
1769
Rados::AioCompletion *Rados::aio_create_completion(void *cb_arg, callback_t cb_complete, callback_t cb_safe)
1632
1770
{
1633
 
  RadosClient::AioCompletion *c = ((RadosClient *)client)->aio_create_completion(cb_arg, cb_complete, cb_safe);
 
1771
  RadosClient::AioCompletion *c = client->aio_create_completion(cb_arg, cb_complete, cb_safe);
1634
1772
  return new AioCompletion(c);
1635
1773
}
1636
1774
 
1706
1844
  if (!rados_initialized) {
1707
1845
    __rados_init(argc, argv);
1708
1846
    radosp = new RadosClient;
1709
 
    radosp->init();
 
1847
    if (!radosp) {
 
1848
      ret = -ENOMEM;
 
1849
      goto done;
 
1850
    }
 
1851
    ret = radosp->init();
 
1852
    if (ret < 0)
 
1853
      goto done;
1710
1854
  }
1711
1855
  ++rados_initialized;
1712
 
 
 
1856
done:
1713
1857
  rados_init_mutex.Unlock();
1714
1858
  return ret;
1715
1859
}
1719
1863
  rados_init_mutex.Lock();
1720
1864
  if (!rados_initialized) {
1721
1865
    dout(0) << "rados_deinitialize() called without rados_initialize()" << dendl;
 
1866
    rados_init_mutex.Unlock();
1722
1867
    return;
1723
1868
  }
1724
1869
  --rados_initialized;
1774
1919
  stats->num_object_clones = r.num_object_clones;
1775
1920
  stats->num_object_copies = r.num_object_copies;
1776
1921
  stats->num_objects_missing_on_primary = r.num_objects_missing_on_primary;
 
1922
  stats->num_objects_unfound = r.num_objects_unfound;
1777
1923
  stats->num_objects_degraded = r.num_objects_degraded;
1778
1924
  stats->num_rd = r.num_rd;
1779
1925
  stats->num_rd_kb = r.num_rd_kb;
1837
1983
  RadosClient::PoolCtx *ctx = (RadosClient::PoolCtx *)pool;
1838
1984
  int ret;
1839
1985
  object_t oid(o);
 
1986
 
1840
1987
  bufferlist bl;
 
1988
  bufferptr bp = buffer::create_static(len, buf);
 
1989
  bl.push_back(bp);
 
1990
  
1841
1991
  ret = radosp->read(*ctx, oid, off, bl, len);
1842
1992
  if (ret >= 0) {
1843
1993
    if (bl.length() > len)
1844
1994
      return -ERANGE;
1845
 
    bl.copy(0, bl.length(), buf);
 
1995
    if (bl.c_str() != buf)
 
1996
      bl.copy(0, bl.length(), buf);
1846
1997
    ret = bl.length();    // hrm :/
1847
1998
  }
1848
1999
 
1982
2133
  return radosp->setxattr(*ctx, oid, name, bl);
1983
2134
}
1984
2135
 
 
2136
extern "C" int rados_rmxattr(rados_pool_t pool, const char *o, const char *name)
 
2137
{
 
2138
  RadosClient::PoolCtx *ctx = (RadosClient::PoolCtx *)pool;
 
2139
  object_t oid(o);
 
2140
  return radosp->rmxattr(*ctx, oid, name);
 
2141
}
 
2142
 
1985
2143
extern "C" int rados_stat(rados_pool_t pool, const char *o, uint64_t *psize, time_t *pmtime)
1986
2144
{
1987
2145
  RadosClient::PoolCtx *ctx = (RadosClient::PoolCtx *)pool;