osd_client.c 39 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539
  1. #include "ceph_debug.h"
  2. #include <linux/err.h>
  3. #include <linux/highmem.h>
  4. #include <linux/mm.h>
  5. #include <linux/pagemap.h>
  6. #include <linux/slab.h>
  7. #include <linux/uaccess.h>
  8. #include "super.h"
  9. #include "osd_client.h"
  10. #include "messenger.h"
  11. #include "decode.h"
  12. #include "auth.h"
  13. #define OSD_REPLY_RESERVE_FRONT_LEN 512
  14. const static struct ceph_connection_operations osd_con_ops;
  15. static void kick_requests(struct ceph_osd_client *osdc, struct ceph_osd *osd);
  16. /*
  17. * Implement client access to distributed object storage cluster.
  18. *
  19. * All data objects are stored within a cluster/cloud of OSDs, or
  20. * "object storage devices." (Note that Ceph OSDs have _nothing_ to
  21. * do with the T10 OSD extensions to SCSI.) Ceph OSDs are simply
  22. * remote daemons serving up and coordinating consistent and safe
  23. * access to storage.
  24. *
  25. * Cluster membership and the mapping of data objects onto storage devices
  26. * are described by the osd map.
  27. *
  28. * We keep track of pending OSD requests (read, write), resubmit
  29. * requests to different OSDs when the cluster topology/data layout
  30. * change, or retry the affected requests when the communications
  31. * channel with an OSD is reset.
  32. */
  33. /*
  34. * calculate the mapping of a file extent onto an object, and fill out the
  35. * request accordingly. shorten extent as necessary if it crosses an
  36. * object boundary.
  37. *
  38. * fill osd op in request message.
  39. */
  40. static void calc_layout(struct ceph_osd_client *osdc,
  41. struct ceph_vino vino, struct ceph_file_layout *layout,
  42. u64 off, u64 *plen,
  43. struct ceph_osd_request *req)
  44. {
  45. struct ceph_osd_request_head *reqhead = req->r_request->front.iov_base;
  46. struct ceph_osd_op *op = (void *)(reqhead + 1);
  47. u64 orig_len = *plen;
  48. u64 objoff, objlen; /* extent in object */
  49. u64 bno;
  50. reqhead->snapid = cpu_to_le64(vino.snap);
  51. /* object extent? */
  52. ceph_calc_file_object_mapping(layout, off, plen, &bno,
  53. &objoff, &objlen);
  54. if (*plen < orig_len)
  55. dout(" skipping last %llu, final file extent %llu~%llu\n",
  56. orig_len - *plen, off, *plen);
  57. sprintf(req->r_oid, "%llx.%08llx", vino.ino, bno);
  58. req->r_oid_len = strlen(req->r_oid);
  59. op->extent.offset = cpu_to_le64(objoff);
  60. op->extent.length = cpu_to_le64(objlen);
  61. req->r_num_pages = calc_pages_for(off, *plen);
  62. dout("calc_layout %s (%d) %llu~%llu (%d pages)\n",
  63. req->r_oid, req->r_oid_len, objoff, objlen, req->r_num_pages);
  64. }
  65. static void remove_replies(struct ceph_osd_request *req)
  66. {
  67. int i;
  68. int max = ARRAY_SIZE(req->replies);
  69. for (i=0; i<max; i++) {
  70. if (req->replies[i])
  71. ceph_msg_put(req->replies[i]);
  72. }
  73. }
  74. /*
  75. * requests
  76. */
  77. void ceph_osdc_release_request(struct kref *kref)
  78. {
  79. struct ceph_osd_request *req = container_of(kref,
  80. struct ceph_osd_request,
  81. r_kref);
  82. if (req->r_request)
  83. ceph_msg_put(req->r_request);
  84. if (req->r_reply)
  85. ceph_msg_put(req->r_reply);
  86. remove_replies(req);
  87. if (req->r_con_filling_msg) {
  88. dout("release_request revoking pages %p from con %p\n",
  89. req->r_pages, req->r_con_filling_msg);
  90. ceph_con_revoke_message(req->r_con_filling_msg,
  91. req->r_reply);
  92. ceph_con_put(req->r_con_filling_msg);
  93. }
  94. if (req->r_own_pages)
  95. ceph_release_page_vector(req->r_pages,
  96. req->r_num_pages);
  97. ceph_put_snap_context(req->r_snapc);
  98. if (req->r_mempool)
  99. mempool_free(req, req->r_osdc->req_mempool);
  100. else
  101. kfree(req);
  102. }
  103. static int alloc_replies(struct ceph_osd_request *req, int num_reply)
  104. {
  105. int i;
  106. int max = ARRAY_SIZE(req->replies);
  107. BUG_ON(num_reply > max);
  108. for (i=0; i<num_reply; i++) {
  109. req->replies[i] = ceph_msg_new(0, OSD_REPLY_RESERVE_FRONT_LEN, 0, 0, NULL);
  110. if (IS_ERR(req->replies[i])) {
  111. int j;
  112. int err = PTR_ERR(req->replies[i]);
  113. for (j = 0; j<=i; j++) {
  114. ceph_msg_put(req->replies[j]);
  115. }
  116. return err;
  117. }
  118. }
  119. for (; i<max; i++) {
  120. req->replies[i] = NULL;
  121. }
  122. req->cur_reply = 0;
  123. return 0;
  124. }
  125. static struct ceph_msg *__get_next_reply(struct ceph_connection *con,
  126. struct ceph_osd_request *req,
  127. int front_len)
  128. {
  129. struct ceph_msg *reply;
  130. if (req->r_con_filling_msg) {
  131. dout("revoking reply msg %p from old con %p\n", req->r_reply,
  132. req->r_con_filling_msg);
  133. ceph_con_revoke_message(req->r_con_filling_msg, req->r_reply);
  134. ceph_con_put(req->r_con_filling_msg);
  135. req->cur_reply = 0;
  136. }
  137. reply = req->replies[req->cur_reply];
  138. if (!reply || front_len > OSD_REPLY_RESERVE_FRONT_LEN) {
  139. /* maybe we can allocate it now? */
  140. reply = ceph_msg_new(0, front_len, 0, 0, NULL);
  141. if (!reply || IS_ERR(reply)) {
  142. pr_err(" reply alloc failed, front_len=%d\n", front_len);
  143. return ERR_PTR(-ENOMEM);
  144. }
  145. }
  146. req->r_con_filling_msg = ceph_con_get(con);
  147. req->r_reply = ceph_msg_get(reply); /* for duration of read over socket */
  148. return ceph_msg_get(reply);
  149. }
  150. /*
  151. * build new request AND message, calculate layout, and adjust file
  152. * extent as needed.
  153. *
  154. * if the file was recently truncated, we include information about its
  155. * old and new size so that the object can be updated appropriately. (we
  156. * avoid synchronously deleting truncated objects because it's slow.)
  157. *
  158. * if @do_sync, include a 'startsync' command so that the osd will flush
  159. * data quickly.
  160. */
  161. struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *osdc,
  162. struct ceph_file_layout *layout,
  163. struct ceph_vino vino,
  164. u64 off, u64 *plen,
  165. int opcode, int flags,
  166. struct ceph_snap_context *snapc,
  167. int do_sync,
  168. u32 truncate_seq,
  169. u64 truncate_size,
  170. struct timespec *mtime,
  171. bool use_mempool, int num_reply)
  172. {
  173. struct ceph_osd_request *req;
  174. struct ceph_msg *msg;
  175. struct ceph_osd_request_head *head;
  176. struct ceph_osd_op *op;
  177. void *p;
  178. int num_op = 1 + do_sync;
  179. size_t msg_size = sizeof(*head) + num_op*sizeof(*op);
  180. int err, i;
  181. if (use_mempool) {
  182. req = mempool_alloc(osdc->req_mempool, GFP_NOFS);
  183. memset(req, 0, sizeof(*req));
  184. } else {
  185. req = kzalloc(sizeof(*req), GFP_NOFS);
  186. }
  187. if (req == NULL)
  188. return ERR_PTR(-ENOMEM);
  189. err = alloc_replies(req, num_reply);
  190. if (err) {
  191. ceph_osdc_put_request(req);
  192. return ERR_PTR(-ENOMEM);
  193. }
  194. req->r_num_prealloc_reply = num_reply;
  195. req->r_osdc = osdc;
  196. req->r_mempool = use_mempool;
  197. kref_init(&req->r_kref);
  198. init_completion(&req->r_completion);
  199. init_completion(&req->r_safe_completion);
  200. INIT_LIST_HEAD(&req->r_unsafe_item);
  201. req->r_flags = flags;
  202. WARN_ON((flags & (CEPH_OSD_FLAG_READ|CEPH_OSD_FLAG_WRITE)) == 0);
  203. /* create message; allow space for oid */
  204. msg_size += 40;
  205. if (snapc)
  206. msg_size += sizeof(u64) * snapc->num_snaps;
  207. if (use_mempool)
  208. msg = ceph_msgpool_get(&osdc->msgpool_op, 0);
  209. else
  210. msg = ceph_msg_new(CEPH_MSG_OSD_OP, msg_size, 0, 0, NULL);
  211. if (IS_ERR(msg)) {
  212. ceph_osdc_put_request(req);
  213. return ERR_PTR(PTR_ERR(msg));
  214. }
  215. msg->hdr.type = cpu_to_le16(CEPH_MSG_OSD_OP);
  216. memset(msg->front.iov_base, 0, msg->front.iov_len);
  217. head = msg->front.iov_base;
  218. op = (void *)(head + 1);
  219. p = (void *)(op + num_op);
  220. req->r_request = msg;
  221. req->r_snapc = ceph_get_snap_context(snapc);
  222. head->client_inc = cpu_to_le32(1); /* always, for now. */
  223. head->flags = cpu_to_le32(flags);
  224. if (flags & CEPH_OSD_FLAG_WRITE)
  225. ceph_encode_timespec(&head->mtime, mtime);
  226. head->num_ops = cpu_to_le16(num_op);
  227. op->op = cpu_to_le16(opcode);
  228. /* calculate max write size */
  229. calc_layout(osdc, vino, layout, off, plen, req);
  230. req->r_file_layout = *layout; /* keep a copy */
  231. if (flags & CEPH_OSD_FLAG_WRITE) {
  232. req->r_request->hdr.data_off = cpu_to_le16(off);
  233. req->r_request->hdr.data_len = cpu_to_le32(*plen);
  234. op->payload_len = cpu_to_le32(*plen);
  235. }
  236. op->extent.truncate_size = cpu_to_le64(truncate_size);
  237. op->extent.truncate_seq = cpu_to_le32(truncate_seq);
  238. /* fill in oid */
  239. head->object_len = cpu_to_le32(req->r_oid_len);
  240. memcpy(p, req->r_oid, req->r_oid_len);
  241. p += req->r_oid_len;
  242. if (do_sync) {
  243. op++;
  244. op->op = cpu_to_le16(CEPH_OSD_OP_STARTSYNC);
  245. }
  246. if (snapc) {
  247. head->snap_seq = cpu_to_le64(snapc->seq);
  248. head->num_snaps = cpu_to_le32(snapc->num_snaps);
  249. for (i = 0; i < snapc->num_snaps; i++) {
  250. put_unaligned_le64(snapc->snaps[i], p);
  251. p += sizeof(u64);
  252. }
  253. }
  254. BUG_ON(p > msg->front.iov_base + msg->front.iov_len);
  255. return req;
  256. }
  257. /*
  258. * We keep osd requests in an rbtree, sorted by ->r_tid.
  259. */
  260. static void __insert_request(struct ceph_osd_client *osdc,
  261. struct ceph_osd_request *new)
  262. {
  263. struct rb_node **p = &osdc->requests.rb_node;
  264. struct rb_node *parent = NULL;
  265. struct ceph_osd_request *req = NULL;
  266. while (*p) {
  267. parent = *p;
  268. req = rb_entry(parent, struct ceph_osd_request, r_node);
  269. if (new->r_tid < req->r_tid)
  270. p = &(*p)->rb_left;
  271. else if (new->r_tid > req->r_tid)
  272. p = &(*p)->rb_right;
  273. else
  274. BUG();
  275. }
  276. rb_link_node(&new->r_node, parent, p);
  277. rb_insert_color(&new->r_node, &osdc->requests);
  278. }
  279. static struct ceph_osd_request *__lookup_request(struct ceph_osd_client *osdc,
  280. u64 tid)
  281. {
  282. struct ceph_osd_request *req;
  283. struct rb_node *n = osdc->requests.rb_node;
  284. while (n) {
  285. req = rb_entry(n, struct ceph_osd_request, r_node);
  286. if (tid < req->r_tid)
  287. n = n->rb_left;
  288. else if (tid > req->r_tid)
  289. n = n->rb_right;
  290. else
  291. return req;
  292. }
  293. return NULL;
  294. }
  295. static struct ceph_osd_request *
  296. __lookup_request_ge(struct ceph_osd_client *osdc,
  297. u64 tid)
  298. {
  299. struct ceph_osd_request *req;
  300. struct rb_node *n = osdc->requests.rb_node;
  301. while (n) {
  302. req = rb_entry(n, struct ceph_osd_request, r_node);
  303. if (tid < req->r_tid) {
  304. if (!n->rb_left)
  305. return req;
  306. n = n->rb_left;
  307. } else if (tid > req->r_tid) {
  308. n = n->rb_right;
  309. } else {
  310. return req;
  311. }
  312. }
  313. return NULL;
  314. }
  315. /*
  316. * If the osd connection drops, we need to resubmit all requests.
  317. */
  318. static void osd_reset(struct ceph_connection *con)
  319. {
  320. struct ceph_osd *osd = con->private;
  321. struct ceph_osd_client *osdc;
  322. if (!osd)
  323. return;
  324. dout("osd_reset osd%d\n", osd->o_osd);
  325. osdc = osd->o_osdc;
  326. down_read(&osdc->map_sem);
  327. kick_requests(osdc, osd);
  328. up_read(&osdc->map_sem);
  329. }
  330. /*
  331. * Track open sessions with osds.
  332. */
  333. static struct ceph_osd *create_osd(struct ceph_osd_client *osdc)
  334. {
  335. struct ceph_osd *osd;
  336. osd = kzalloc(sizeof(*osd), GFP_NOFS);
  337. if (!osd)
  338. return NULL;
  339. atomic_set(&osd->o_ref, 1);
  340. osd->o_osdc = osdc;
  341. INIT_LIST_HEAD(&osd->o_requests);
  342. INIT_LIST_HEAD(&osd->o_osd_lru);
  343. osd->o_incarnation = 1;
  344. ceph_con_init(osdc->client->msgr, &osd->o_con);
  345. osd->o_con.private = osd;
  346. osd->o_con.ops = &osd_con_ops;
  347. osd->o_con.peer_name.type = CEPH_ENTITY_TYPE_OSD;
  348. return osd;
  349. }
  350. static struct ceph_osd *get_osd(struct ceph_osd *osd)
  351. {
  352. if (atomic_inc_not_zero(&osd->o_ref)) {
  353. dout("get_osd %p %d -> %d\n", osd, atomic_read(&osd->o_ref)-1,
  354. atomic_read(&osd->o_ref));
  355. return osd;
  356. } else {
  357. dout("get_osd %p FAIL\n", osd);
  358. return NULL;
  359. }
  360. }
  361. static void put_osd(struct ceph_osd *osd)
  362. {
  363. dout("put_osd %p %d -> %d\n", osd, atomic_read(&osd->o_ref),
  364. atomic_read(&osd->o_ref) - 1);
  365. if (atomic_dec_and_test(&osd->o_ref))
  366. kfree(osd);
  367. }
  368. /*
  369. * remove an osd from our map
  370. */
  371. static void __remove_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd)
  372. {
  373. dout("__remove_osd %p\n", osd);
  374. BUG_ON(!list_empty(&osd->o_requests));
  375. rb_erase(&osd->o_node, &osdc->osds);
  376. list_del_init(&osd->o_osd_lru);
  377. ceph_con_close(&osd->o_con);
  378. put_osd(osd);
  379. }
  380. static void __move_osd_to_lru(struct ceph_osd_client *osdc,
  381. struct ceph_osd *osd)
  382. {
  383. dout("__move_osd_to_lru %p\n", osd);
  384. BUG_ON(!list_empty(&osd->o_osd_lru));
  385. list_add_tail(&osd->o_osd_lru, &osdc->osd_lru);
  386. osd->lru_ttl = jiffies + osdc->client->mount_args->osd_idle_ttl * HZ;
  387. }
  388. static void __remove_osd_from_lru(struct ceph_osd *osd)
  389. {
  390. dout("__remove_osd_from_lru %p\n", osd);
  391. if (!list_empty(&osd->o_osd_lru))
  392. list_del_init(&osd->o_osd_lru);
  393. }
  394. static void remove_old_osds(struct ceph_osd_client *osdc, int remove_all)
  395. {
  396. struct ceph_osd *osd, *nosd;
  397. dout("__remove_old_osds %p\n", osdc);
  398. mutex_lock(&osdc->request_mutex);
  399. list_for_each_entry_safe(osd, nosd, &osdc->osd_lru, o_osd_lru) {
  400. if (!remove_all && time_before(jiffies, osd->lru_ttl))
  401. break;
  402. __remove_osd(osdc, osd);
  403. }
  404. mutex_unlock(&osdc->request_mutex);
  405. }
  406. /*
  407. * reset osd connect
  408. */
  409. static int __reset_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd)
  410. {
  411. int ret = 0;
  412. dout("__reset_osd %p osd%d\n", osd, osd->o_osd);
  413. if (list_empty(&osd->o_requests)) {
  414. __remove_osd(osdc, osd);
  415. } else {
  416. ceph_con_close(&osd->o_con);
  417. ceph_con_open(&osd->o_con, &osdc->osdmap->osd_addr[osd->o_osd]);
  418. osd->o_incarnation++;
  419. }
  420. return ret;
  421. }
  422. static void __insert_osd(struct ceph_osd_client *osdc, struct ceph_osd *new)
  423. {
  424. struct rb_node **p = &osdc->osds.rb_node;
  425. struct rb_node *parent = NULL;
  426. struct ceph_osd *osd = NULL;
  427. while (*p) {
  428. parent = *p;
  429. osd = rb_entry(parent, struct ceph_osd, o_node);
  430. if (new->o_osd < osd->o_osd)
  431. p = &(*p)->rb_left;
  432. else if (new->o_osd > osd->o_osd)
  433. p = &(*p)->rb_right;
  434. else
  435. BUG();
  436. }
  437. rb_link_node(&new->o_node, parent, p);
  438. rb_insert_color(&new->o_node, &osdc->osds);
  439. }
  440. static struct ceph_osd *__lookup_osd(struct ceph_osd_client *osdc, int o)
  441. {
  442. struct ceph_osd *osd;
  443. struct rb_node *n = osdc->osds.rb_node;
  444. while (n) {
  445. osd = rb_entry(n, struct ceph_osd, o_node);
  446. if (o < osd->o_osd)
  447. n = n->rb_left;
  448. else if (o > osd->o_osd)
  449. n = n->rb_right;
  450. else
  451. return osd;
  452. }
  453. return NULL;
  454. }
  455. /*
  456. * Register request, assign tid. If this is the first request, set up
  457. * the timeout event.
  458. */
  459. static void register_request(struct ceph_osd_client *osdc,
  460. struct ceph_osd_request *req)
  461. {
  462. mutex_lock(&osdc->request_mutex);
  463. req->r_tid = ++osdc->last_tid;
  464. req->r_request->hdr.tid = cpu_to_le64(req->r_tid);
  465. dout("register_request %p tid %lld\n", req, req->r_tid);
  466. __insert_request(osdc, req);
  467. ceph_osdc_get_request(req);
  468. osdc->num_requests++;
  469. req->r_timeout_stamp =
  470. jiffies + osdc->client->mount_args->osd_timeout*HZ;
  471. if (osdc->num_requests == 1) {
  472. osdc->timeout_tid = req->r_tid;
  473. dout(" timeout on tid %llu at %lu\n", req->r_tid,
  474. req->r_timeout_stamp);
  475. schedule_delayed_work(&osdc->timeout_work,
  476. round_jiffies_relative(req->r_timeout_stamp - jiffies));
  477. }
  478. mutex_unlock(&osdc->request_mutex);
  479. }
  480. /*
  481. * called under osdc->request_mutex
  482. */
  483. static void __unregister_request(struct ceph_osd_client *osdc,
  484. struct ceph_osd_request *req)
  485. {
  486. dout("__unregister_request %p tid %lld\n", req, req->r_tid);
  487. rb_erase(&req->r_node, &osdc->requests);
  488. osdc->num_requests--;
  489. if (req->r_osd) {
  490. /* make sure the original request isn't in flight. */
  491. ceph_con_revoke(&req->r_osd->o_con, req->r_request);
  492. list_del_init(&req->r_osd_item);
  493. if (list_empty(&req->r_osd->o_requests))
  494. __move_osd_to_lru(osdc, req->r_osd);
  495. req->r_osd = NULL;
  496. }
  497. ceph_osdc_put_request(req);
  498. if (req->r_tid == osdc->timeout_tid) {
  499. if (osdc->num_requests == 0) {
  500. dout("no requests, canceling timeout\n");
  501. osdc->timeout_tid = 0;
  502. cancel_delayed_work(&osdc->timeout_work);
  503. } else {
  504. req = rb_entry(rb_first(&osdc->requests),
  505. struct ceph_osd_request, r_node);
  506. osdc->timeout_tid = req->r_tid;
  507. dout("rescheduled timeout on tid %llu at %lu\n",
  508. req->r_tid, req->r_timeout_stamp);
  509. schedule_delayed_work(&osdc->timeout_work,
  510. round_jiffies_relative(req->r_timeout_stamp -
  511. jiffies));
  512. }
  513. }
  514. }
  515. /*
  516. * Cancel a previously queued request message
  517. */
  518. static void __cancel_request(struct ceph_osd_request *req)
  519. {
  520. if (req->r_sent) {
  521. ceph_con_revoke(&req->r_osd->o_con, req->r_request);
  522. req->r_sent = 0;
  523. }
  524. }
  525. /*
  526. * Pick an osd (the first 'up' osd in the pg), allocate the osd struct
  527. * (as needed), and set the request r_osd appropriately. If there is
  528. * no up osd, set r_osd to NULL.
  529. *
  530. * Return 0 if unchanged, 1 if changed, or negative on error.
  531. *
  532. * Caller should hold map_sem for read and request_mutex.
  533. */
  534. static int __map_osds(struct ceph_osd_client *osdc,
  535. struct ceph_osd_request *req)
  536. {
  537. struct ceph_osd_request_head *reqhead = req->r_request->front.iov_base;
  538. struct ceph_pg pgid;
  539. int o = -1;
  540. int err;
  541. dout("map_osds %p tid %lld\n", req, req->r_tid);
  542. err = ceph_calc_object_layout(&reqhead->layout, req->r_oid,
  543. &req->r_file_layout, osdc->osdmap);
  544. if (err)
  545. return err;
  546. pgid = reqhead->layout.ol_pgid;
  547. req->r_pgid = pgid;
  548. o = ceph_calc_pg_primary(osdc->osdmap, pgid);
  549. if ((req->r_osd && req->r_osd->o_osd == o &&
  550. req->r_sent >= req->r_osd->o_incarnation) ||
  551. (req->r_osd == NULL && o == -1))
  552. return 0; /* no change */
  553. dout("map_osds tid %llu pgid %d.%x osd%d (was osd%d)\n",
  554. req->r_tid, le32_to_cpu(pgid.pool), le16_to_cpu(pgid.ps), o,
  555. req->r_osd ? req->r_osd->o_osd : -1);
  556. if (req->r_osd) {
  557. __cancel_request(req);
  558. list_del_init(&req->r_osd_item);
  559. req->r_osd = NULL;
  560. }
  561. req->r_osd = __lookup_osd(osdc, o);
  562. if (!req->r_osd && o >= 0) {
  563. err = -ENOMEM;
  564. req->r_osd = create_osd(osdc);
  565. if (!req->r_osd)
  566. goto out;
  567. dout("map_osds osd %p is osd%d\n", req->r_osd, o);
  568. req->r_osd->o_osd = o;
  569. req->r_osd->o_con.peer_name.num = cpu_to_le64(o);
  570. __insert_osd(osdc, req->r_osd);
  571. ceph_con_open(&req->r_osd->o_con, &osdc->osdmap->osd_addr[o]);
  572. }
  573. if (req->r_osd) {
  574. __remove_osd_from_lru(req->r_osd);
  575. list_add(&req->r_osd_item, &req->r_osd->o_requests);
  576. }
  577. err = 1; /* osd changed */
  578. out:
  579. return err;
  580. }
  581. /*
  582. * caller should hold map_sem (for read) and request_mutex
  583. */
  584. static int __send_request(struct ceph_osd_client *osdc,
  585. struct ceph_osd_request *req)
  586. {
  587. struct ceph_osd_request_head *reqhead;
  588. int err;
  589. err = __map_osds(osdc, req);
  590. if (err < 0)
  591. return err;
  592. if (req->r_osd == NULL) {
  593. dout("send_request %p no up osds in pg\n", req);
  594. ceph_monc_request_next_osdmap(&osdc->client->monc);
  595. return 0;
  596. }
  597. dout("send_request %p tid %llu to osd%d flags %d\n",
  598. req, req->r_tid, req->r_osd->o_osd, req->r_flags);
  599. reqhead = req->r_request->front.iov_base;
  600. reqhead->osdmap_epoch = cpu_to_le32(osdc->osdmap->epoch);
  601. reqhead->flags |= cpu_to_le32(req->r_flags); /* e.g., RETRY */
  602. reqhead->reassert_version = req->r_reassert_version;
  603. req->r_timeout_stamp = jiffies+osdc->client->mount_args->osd_timeout*HZ;
  604. ceph_msg_get(req->r_request); /* send consumes a ref */
  605. ceph_con_send(&req->r_osd->o_con, req->r_request);
  606. req->r_sent = req->r_osd->o_incarnation;
  607. return 0;
  608. }
  609. /*
  610. * Timeout callback, called every N seconds when 1 or more osd
  611. * requests has been active for more than N seconds. When this
  612. * happens, we ping all OSDs with requests who have timed out to
  613. * ensure any communications channel reset is detected. Reset the
  614. * request timeouts another N seconds in the future as we go.
  615. * Reschedule the timeout event another N seconds in future (unless
  616. * there are no open requests).
  617. */
  618. static void handle_timeout(struct work_struct *work)
  619. {
  620. struct ceph_osd_client *osdc =
  621. container_of(work, struct ceph_osd_client, timeout_work.work);
  622. struct ceph_osd_request *req;
  623. struct ceph_osd *osd;
  624. unsigned long timeout = osdc->client->mount_args->osd_timeout * HZ;
  625. unsigned long next_timeout = timeout + jiffies;
  626. struct rb_node *p;
  627. dout("timeout\n");
  628. down_read(&osdc->map_sem);
  629. ceph_monc_request_next_osdmap(&osdc->client->monc);
  630. mutex_lock(&osdc->request_mutex);
  631. for (p = rb_first(&osdc->requests); p; p = rb_next(p)) {
  632. req = rb_entry(p, struct ceph_osd_request, r_node);
  633. if (req->r_resend) {
  634. int err;
  635. dout("osdc resending prev failed %lld\n", req->r_tid);
  636. err = __send_request(osdc, req);
  637. if (err)
  638. dout("osdc failed again on %lld\n", req->r_tid);
  639. else
  640. req->r_resend = false;
  641. continue;
  642. }
  643. }
  644. for (p = rb_first(&osdc->osds); p; p = rb_next(p)) {
  645. osd = rb_entry(p, struct ceph_osd, o_node);
  646. if (list_empty(&osd->o_requests))
  647. continue;
  648. req = list_first_entry(&osd->o_requests,
  649. struct ceph_osd_request, r_osd_item);
  650. if (time_before(jiffies, req->r_timeout_stamp))
  651. continue;
  652. dout(" tid %llu (at least) timed out on osd%d\n",
  653. req->r_tid, osd->o_osd);
  654. req->r_timeout_stamp = next_timeout;
  655. ceph_con_keepalive(&osd->o_con);
  656. }
  657. if (osdc->timeout_tid)
  658. schedule_delayed_work(&osdc->timeout_work,
  659. round_jiffies_relative(timeout));
  660. mutex_unlock(&osdc->request_mutex);
  661. up_read(&osdc->map_sem);
  662. }
  663. static void handle_osds_timeout(struct work_struct *work)
  664. {
  665. struct ceph_osd_client *osdc =
  666. container_of(work, struct ceph_osd_client,
  667. osds_timeout_work.work);
  668. unsigned long delay =
  669. osdc->client->mount_args->osd_idle_ttl * HZ >> 2;
  670. dout("osds timeout\n");
  671. down_read(&osdc->map_sem);
  672. remove_old_osds(osdc, 0);
  673. up_read(&osdc->map_sem);
  674. schedule_delayed_work(&osdc->osds_timeout_work,
  675. round_jiffies_relative(delay));
  676. }
  677. /*
  678. * handle osd op reply. either call the callback if it is specified,
  679. * or do the completion to wake up the waiting thread.
  680. */
  681. static void handle_reply(struct ceph_osd_client *osdc, struct ceph_msg *msg,
  682. struct ceph_connection *con)
  683. {
  684. struct ceph_osd_reply_head *rhead = msg->front.iov_base;
  685. struct ceph_osd_request *req;
  686. u64 tid;
  687. int numops, object_len, flags;
  688. tid = le64_to_cpu(msg->hdr.tid);
  689. if (msg->front.iov_len < sizeof(*rhead))
  690. goto bad;
  691. numops = le32_to_cpu(rhead->num_ops);
  692. object_len = le32_to_cpu(rhead->object_len);
  693. if (msg->front.iov_len != sizeof(*rhead) + object_len +
  694. numops * sizeof(struct ceph_osd_op))
  695. goto bad;
  696. dout("handle_reply %p tid %llu\n", msg, tid);
  697. /* lookup */
  698. mutex_lock(&osdc->request_mutex);
  699. req = __lookup_request(osdc, tid);
  700. if (req == NULL) {
  701. dout("handle_reply tid %llu dne\n", tid);
  702. mutex_unlock(&osdc->request_mutex);
  703. return;
  704. }
  705. ceph_osdc_get_request(req);
  706. flags = le32_to_cpu(rhead->flags);
  707. /*
  708. * if this connection filled our message, drop our reference now, to
  709. * avoid a (safe but slower) revoke later.
  710. */
  711. if (req->r_con_filling_msg == con && req->r_reply == msg) {
  712. dout(" got pages, dropping con_filling_msg ref %p\n", con);
  713. req->r_con_filling_msg = NULL;
  714. ceph_con_put(con);
  715. }
  716. if (req->r_reply) {
  717. /*
  718. * once we see the message has been received, we don't
  719. * need a ref (which is only needed for revoking
  720. * pages)
  721. */
  722. ceph_msg_put(req->r_reply);
  723. req->r_reply = NULL;
  724. }
  725. if (!req->r_got_reply) {
  726. unsigned bytes;
  727. req->r_result = le32_to_cpu(rhead->result);
  728. bytes = le32_to_cpu(msg->hdr.data_len);
  729. dout("handle_reply result %d bytes %d\n", req->r_result,
  730. bytes);
  731. if (req->r_result == 0)
  732. req->r_result = bytes;
  733. /* in case this is a write and we need to replay, */
  734. req->r_reassert_version = rhead->reassert_version;
  735. req->r_got_reply = 1;
  736. } else if ((flags & CEPH_OSD_FLAG_ONDISK) == 0) {
  737. dout("handle_reply tid %llu dup ack\n", tid);
  738. mutex_unlock(&osdc->request_mutex);
  739. goto done;
  740. }
  741. dout("handle_reply tid %llu flags %d\n", tid, flags);
  742. /* either this is a read, or we got the safe response */
  743. if ((flags & CEPH_OSD_FLAG_ONDISK) ||
  744. ((flags & CEPH_OSD_FLAG_WRITE) == 0))
  745. __unregister_request(osdc, req);
  746. mutex_unlock(&osdc->request_mutex);
  747. if (req->r_callback)
  748. req->r_callback(req, msg);
  749. else
  750. complete(&req->r_completion);
  751. if (flags & CEPH_OSD_FLAG_ONDISK) {
  752. if (req->r_safe_callback)
  753. req->r_safe_callback(req, msg);
  754. complete(&req->r_safe_completion); /* fsync waiter */
  755. }
  756. done:
  757. ceph_osdc_put_request(req);
  758. return;
  759. bad:
  760. pr_err("corrupt osd_op_reply got %d %d expected %d\n",
  761. (int)msg->front.iov_len, le32_to_cpu(msg->hdr.front_len),
  762. (int)sizeof(*rhead));
  763. ceph_msg_dump(msg);
  764. }
  765. /*
  766. * Resubmit osd requests whose osd or osd address has changed. Request
  767. * a new osd map if osds are down, or we are otherwise unable to determine
  768. * how to direct a request.
  769. *
  770. * Close connections to down osds.
  771. *
  772. * If @who is specified, resubmit requests for that specific osd.
  773. *
  774. * Caller should hold map_sem for read and request_mutex.
  775. */
  776. static void kick_requests(struct ceph_osd_client *osdc,
  777. struct ceph_osd *kickosd)
  778. {
  779. struct ceph_osd_request *req;
  780. struct rb_node *p, *n;
  781. int needmap = 0;
  782. int err;
  783. dout("kick_requests osd%d\n", kickosd ? kickosd->o_osd : -1);
  784. mutex_lock(&osdc->request_mutex);
  785. if (kickosd) {
  786. __reset_osd(osdc, kickosd);
  787. } else {
  788. for (p = rb_first(&osdc->osds); p; p = n) {
  789. struct ceph_osd *osd =
  790. rb_entry(p, struct ceph_osd, o_node);
  791. n = rb_next(p);
  792. if (!ceph_osd_is_up(osdc->osdmap, osd->o_osd) ||
  793. memcmp(&osd->o_con.peer_addr,
  794. ceph_osd_addr(osdc->osdmap,
  795. osd->o_osd),
  796. sizeof(struct ceph_entity_addr)) != 0)
  797. __reset_osd(osdc, osd);
  798. }
  799. }
  800. for (p = rb_first(&osdc->requests); p; p = rb_next(p)) {
  801. req = rb_entry(p, struct ceph_osd_request, r_node);
  802. if (req->r_resend) {
  803. dout(" r_resend set on tid %llu\n", req->r_tid);
  804. __cancel_request(req);
  805. goto kick;
  806. }
  807. if (req->r_osd && kickosd == req->r_osd) {
  808. __cancel_request(req);
  809. goto kick;
  810. }
  811. err = __map_osds(osdc, req);
  812. if (err == 0)
  813. continue; /* no change */
  814. if (err < 0) {
  815. /*
  816. * FIXME: really, we should set the request
  817. * error and fail if this isn't a 'nofail'
  818. * request, but that's a fair bit more
  819. * complicated to do. So retry!
  820. */
  821. dout(" setting r_resend on %llu\n", req->r_tid);
  822. req->r_resend = true;
  823. continue;
  824. }
  825. if (req->r_osd == NULL) {
  826. dout("tid %llu maps to no valid osd\n", req->r_tid);
  827. needmap++; /* request a newer map */
  828. continue;
  829. }
  830. kick:
  831. dout("kicking %p tid %llu osd%d\n", req, req->r_tid,
  832. req->r_osd->o_osd);
  833. req->r_flags |= CEPH_OSD_FLAG_RETRY;
  834. err = __send_request(osdc, req);
  835. if (err) {
  836. dout(" setting r_resend on %llu\n", req->r_tid);
  837. req->r_resend = true;
  838. }
  839. }
  840. mutex_unlock(&osdc->request_mutex);
  841. if (needmap) {
  842. dout("%d requests for down osds, need new map\n", needmap);
  843. ceph_monc_request_next_osdmap(&osdc->client->monc);
  844. }
  845. }
  846. /*
  847. * Process updated osd map.
  848. *
  849. * The message contains any number of incremental and full maps, normally
  850. * indicating some sort of topology change in the cluster. Kick requests
  851. * off to different OSDs as needed.
  852. */
  853. void ceph_osdc_handle_map(struct ceph_osd_client *osdc, struct ceph_msg *msg)
  854. {
  855. void *p, *end, *next;
  856. u32 nr_maps, maplen;
  857. u32 epoch;
  858. struct ceph_osdmap *newmap = NULL, *oldmap;
  859. int err;
  860. struct ceph_fsid fsid;
  861. dout("handle_map have %u\n", osdc->osdmap ? osdc->osdmap->epoch : 0);
  862. p = msg->front.iov_base;
  863. end = p + msg->front.iov_len;
  864. /* verify fsid */
  865. ceph_decode_need(&p, end, sizeof(fsid), bad);
  866. ceph_decode_copy(&p, &fsid, sizeof(fsid));
  867. if (ceph_check_fsid(osdc->client, &fsid) < 0)
  868. return;
  869. down_write(&osdc->map_sem);
  870. /* incremental maps */
  871. ceph_decode_32_safe(&p, end, nr_maps, bad);
  872. dout(" %d inc maps\n", nr_maps);
  873. while (nr_maps > 0) {
  874. ceph_decode_need(&p, end, 2*sizeof(u32), bad);
  875. epoch = ceph_decode_32(&p);
  876. maplen = ceph_decode_32(&p);
  877. ceph_decode_need(&p, end, maplen, bad);
  878. next = p + maplen;
  879. if (osdc->osdmap && osdc->osdmap->epoch+1 == epoch) {
  880. dout("applying incremental map %u len %d\n",
  881. epoch, maplen);
  882. newmap = osdmap_apply_incremental(&p, next,
  883. osdc->osdmap,
  884. osdc->client->msgr);
  885. if (IS_ERR(newmap)) {
  886. err = PTR_ERR(newmap);
  887. goto bad;
  888. }
  889. BUG_ON(!newmap);
  890. if (newmap != osdc->osdmap) {
  891. ceph_osdmap_destroy(osdc->osdmap);
  892. osdc->osdmap = newmap;
  893. }
  894. } else {
  895. dout("ignoring incremental map %u len %d\n",
  896. epoch, maplen);
  897. }
  898. p = next;
  899. nr_maps--;
  900. }
  901. if (newmap)
  902. goto done;
  903. /* full maps */
  904. ceph_decode_32_safe(&p, end, nr_maps, bad);
  905. dout(" %d full maps\n", nr_maps);
  906. while (nr_maps) {
  907. ceph_decode_need(&p, end, 2*sizeof(u32), bad);
  908. epoch = ceph_decode_32(&p);
  909. maplen = ceph_decode_32(&p);
  910. ceph_decode_need(&p, end, maplen, bad);
  911. if (nr_maps > 1) {
  912. dout("skipping non-latest full map %u len %d\n",
  913. epoch, maplen);
  914. } else if (osdc->osdmap && osdc->osdmap->epoch >= epoch) {
  915. dout("skipping full map %u len %d, "
  916. "older than our %u\n", epoch, maplen,
  917. osdc->osdmap->epoch);
  918. } else {
  919. dout("taking full map %u len %d\n", epoch, maplen);
  920. newmap = osdmap_decode(&p, p+maplen);
  921. if (IS_ERR(newmap)) {
  922. err = PTR_ERR(newmap);
  923. goto bad;
  924. }
  925. BUG_ON(!newmap);
  926. oldmap = osdc->osdmap;
  927. osdc->osdmap = newmap;
  928. if (oldmap)
  929. ceph_osdmap_destroy(oldmap);
  930. }
  931. p += maplen;
  932. nr_maps--;
  933. }
  934. done:
  935. downgrade_write(&osdc->map_sem);
  936. ceph_monc_got_osdmap(&osdc->client->monc, osdc->osdmap->epoch);
  937. if (newmap)
  938. kick_requests(osdc, NULL);
  939. up_read(&osdc->map_sem);
  940. return;
  941. bad:
  942. pr_err("osdc handle_map corrupt msg\n");
  943. ceph_msg_dump(msg);
  944. up_write(&osdc->map_sem);
  945. return;
  946. }
  947. /*
  948. * A read request prepares specific pages that data is to be read into.
  949. * When a message is being read off the wire, we call prepare_pages to
  950. * find those pages.
  951. * 0 = success, -1 failure.
  952. */
  953. static int __prepare_pages(struct ceph_connection *con,
  954. struct ceph_msg_header *hdr,
  955. struct ceph_osd_request *req,
  956. u64 tid,
  957. struct ceph_msg *m)
  958. {
  959. struct ceph_osd *osd = con->private;
  960. struct ceph_osd_client *osdc;
  961. int ret = -1;
  962. int data_len = le32_to_cpu(hdr->data_len);
  963. unsigned data_off = le16_to_cpu(hdr->data_off);
  964. int want = calc_pages_for(data_off & ~PAGE_MASK, data_len);
  965. if (!osd)
  966. return -1;
  967. osdc = osd->o_osdc;
  968. dout("__prepare_pages on msg %p tid %llu, has %d pages, want %d\n", m,
  969. tid, req->r_num_pages, want);
  970. if (unlikely(req->r_num_pages < want))
  971. goto out;
  972. m->pages = req->r_pages;
  973. m->nr_pages = req->r_num_pages;
  974. ret = 0; /* success */
  975. out:
  976. BUG_ON(ret < 0 || m->nr_pages < want);
  977. return ret;
  978. }
  979. /*
  980. * Register request, send initial attempt.
  981. */
  982. int ceph_osdc_start_request(struct ceph_osd_client *osdc,
  983. struct ceph_osd_request *req,
  984. bool nofail)
  985. {
  986. int rc = 0;
  987. req->r_request->pages = req->r_pages;
  988. req->r_request->nr_pages = req->r_num_pages;
  989. register_request(osdc, req);
  990. down_read(&osdc->map_sem);
  991. mutex_lock(&osdc->request_mutex);
  992. /*
  993. * a racing kick_requests() may have sent the message for us
  994. * while we dropped request_mutex above, so only send now if
  995. * the request still han't been touched yet.
  996. */
  997. if (req->r_sent == 0) {
  998. rc = __send_request(osdc, req);
  999. if (rc) {
  1000. if (nofail) {
  1001. dout("osdc_start_request failed send, "
  1002. " marking %lld\n", req->r_tid);
  1003. req->r_resend = true;
  1004. rc = 0;
  1005. } else {
  1006. __unregister_request(osdc, req);
  1007. }
  1008. }
  1009. }
  1010. mutex_unlock(&osdc->request_mutex);
  1011. up_read(&osdc->map_sem);
  1012. return rc;
  1013. }
  1014. /*
  1015. * wait for a request to complete
  1016. */
  1017. int ceph_osdc_wait_request(struct ceph_osd_client *osdc,
  1018. struct ceph_osd_request *req)
  1019. {
  1020. int rc;
  1021. rc = wait_for_completion_interruptible(&req->r_completion);
  1022. if (rc < 0) {
  1023. mutex_lock(&osdc->request_mutex);
  1024. __cancel_request(req);
  1025. __unregister_request(osdc, req);
  1026. mutex_unlock(&osdc->request_mutex);
  1027. dout("wait_request tid %llu canceled/timed out\n", req->r_tid);
  1028. return rc;
  1029. }
  1030. dout("wait_request tid %llu result %d\n", req->r_tid, req->r_result);
  1031. return req->r_result;
  1032. }
  1033. /*
  1034. * sync - wait for all in-flight requests to flush. avoid starvation.
  1035. */
  1036. void ceph_osdc_sync(struct ceph_osd_client *osdc)
  1037. {
  1038. struct ceph_osd_request *req;
  1039. u64 last_tid, next_tid = 0;
  1040. mutex_lock(&osdc->request_mutex);
  1041. last_tid = osdc->last_tid;
  1042. while (1) {
  1043. req = __lookup_request_ge(osdc, next_tid);
  1044. if (!req)
  1045. break;
  1046. if (req->r_tid > last_tid)
  1047. break;
  1048. next_tid = req->r_tid + 1;
  1049. if ((req->r_flags & CEPH_OSD_FLAG_WRITE) == 0)
  1050. continue;
  1051. ceph_osdc_get_request(req);
  1052. mutex_unlock(&osdc->request_mutex);
  1053. dout("sync waiting on tid %llu (last is %llu)\n",
  1054. req->r_tid, last_tid);
  1055. wait_for_completion(&req->r_safe_completion);
  1056. mutex_lock(&osdc->request_mutex);
  1057. ceph_osdc_put_request(req);
  1058. }
  1059. mutex_unlock(&osdc->request_mutex);
  1060. dout("sync done (thru tid %llu)\n", last_tid);
  1061. }
  1062. /*
  1063. * init, shutdown
  1064. */
  1065. int ceph_osdc_init(struct ceph_osd_client *osdc, struct ceph_client *client)
  1066. {
  1067. int err;
  1068. dout("init\n");
  1069. osdc->client = client;
  1070. osdc->osdmap = NULL;
  1071. init_rwsem(&osdc->map_sem);
  1072. init_completion(&osdc->map_waiters);
  1073. osdc->last_requested_map = 0;
  1074. mutex_init(&osdc->request_mutex);
  1075. osdc->timeout_tid = 0;
  1076. osdc->last_tid = 0;
  1077. osdc->osds = RB_ROOT;
  1078. INIT_LIST_HEAD(&osdc->osd_lru);
  1079. osdc->requests = RB_ROOT;
  1080. osdc->num_requests = 0;
  1081. INIT_DELAYED_WORK(&osdc->timeout_work, handle_timeout);
  1082. INIT_DELAYED_WORK(&osdc->osds_timeout_work, handle_osds_timeout);
  1083. schedule_delayed_work(&osdc->osds_timeout_work,
  1084. round_jiffies_relative(osdc->client->mount_args->osd_idle_ttl * HZ));
  1085. err = -ENOMEM;
  1086. osdc->req_mempool = mempool_create_kmalloc_pool(10,
  1087. sizeof(struct ceph_osd_request));
  1088. if (!osdc->req_mempool)
  1089. goto out;
  1090. err = ceph_msgpool_init(&osdc->msgpool_op, 4096, 10, true);
  1091. if (err < 0)
  1092. goto out_mempool;
  1093. return 0;
  1094. out_mempool:
  1095. mempool_destroy(osdc->req_mempool);
  1096. out:
  1097. return err;
  1098. }
  1099. void ceph_osdc_stop(struct ceph_osd_client *osdc)
  1100. {
  1101. cancel_delayed_work_sync(&osdc->timeout_work);
  1102. cancel_delayed_work_sync(&osdc->osds_timeout_work);
  1103. if (osdc->osdmap) {
  1104. ceph_osdmap_destroy(osdc->osdmap);
  1105. osdc->osdmap = NULL;
  1106. }
  1107. remove_old_osds(osdc, 1);
  1108. mempool_destroy(osdc->req_mempool);
  1109. ceph_msgpool_destroy(&osdc->msgpool_op);
  1110. }
  1111. /*
  1112. * Read some contiguous pages. If we cross a stripe boundary, shorten
  1113. * *plen. Return number of bytes read, or error.
  1114. */
  1115. int ceph_osdc_readpages(struct ceph_osd_client *osdc,
  1116. struct ceph_vino vino, struct ceph_file_layout *layout,
  1117. u64 off, u64 *plen,
  1118. u32 truncate_seq, u64 truncate_size,
  1119. struct page **pages, int num_pages)
  1120. {
  1121. struct ceph_osd_request *req;
  1122. int rc = 0;
  1123. dout("readpages on ino %llx.%llx on %llu~%llu\n", vino.ino,
  1124. vino.snap, off, *plen);
  1125. req = ceph_osdc_new_request(osdc, layout, vino, off, plen,
  1126. CEPH_OSD_OP_READ, CEPH_OSD_FLAG_READ,
  1127. NULL, 0, truncate_seq, truncate_size, NULL,
  1128. false, 1);
  1129. if (IS_ERR(req))
  1130. return PTR_ERR(req);
  1131. /* it may be a short read due to an object boundary */
  1132. req->r_pages = pages;
  1133. num_pages = calc_pages_for(off, *plen);
  1134. req->r_num_pages = num_pages;
  1135. dout("readpages final extent is %llu~%llu (%d pages)\n",
  1136. off, *plen, req->r_num_pages);
  1137. rc = ceph_osdc_start_request(osdc, req, false);
  1138. if (!rc)
  1139. rc = ceph_osdc_wait_request(osdc, req);
  1140. ceph_osdc_put_request(req);
  1141. dout("readpages result %d\n", rc);
  1142. return rc;
  1143. }
  1144. /*
  1145. * do a synchronous write on N pages
  1146. */
  1147. int ceph_osdc_writepages(struct ceph_osd_client *osdc, struct ceph_vino vino,
  1148. struct ceph_file_layout *layout,
  1149. struct ceph_snap_context *snapc,
  1150. u64 off, u64 len,
  1151. u32 truncate_seq, u64 truncate_size,
  1152. struct timespec *mtime,
  1153. struct page **pages, int num_pages,
  1154. int flags, int do_sync, bool nofail)
  1155. {
  1156. struct ceph_osd_request *req;
  1157. int rc = 0;
  1158. BUG_ON(vino.snap != CEPH_NOSNAP);
  1159. req = ceph_osdc_new_request(osdc, layout, vino, off, &len,
  1160. CEPH_OSD_OP_WRITE,
  1161. flags | CEPH_OSD_FLAG_ONDISK |
  1162. CEPH_OSD_FLAG_WRITE,
  1163. snapc, do_sync,
  1164. truncate_seq, truncate_size, mtime,
  1165. nofail, 1);
  1166. if (IS_ERR(req))
  1167. return PTR_ERR(req);
  1168. /* it may be a short write due to an object boundary */
  1169. req->r_pages = pages;
  1170. req->r_num_pages = calc_pages_for(off, len);
  1171. dout("writepages %llu~%llu (%d pages)\n", off, len,
  1172. req->r_num_pages);
  1173. rc = ceph_osdc_start_request(osdc, req, nofail);
  1174. if (!rc)
  1175. rc = ceph_osdc_wait_request(osdc, req);
  1176. ceph_osdc_put_request(req);
  1177. if (rc == 0)
  1178. rc = len;
  1179. dout("writepages result %d\n", rc);
  1180. return rc;
  1181. }
  1182. /*
  1183. * handle incoming message
  1184. */
  1185. static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
  1186. {
  1187. struct ceph_osd *osd = con->private;
  1188. struct ceph_osd_client *osdc;
  1189. int type = le16_to_cpu(msg->hdr.type);
  1190. if (!osd)
  1191. return;
  1192. osdc = osd->o_osdc;
  1193. switch (type) {
  1194. case CEPH_MSG_OSD_MAP:
  1195. ceph_osdc_handle_map(osdc, msg);
  1196. break;
  1197. case CEPH_MSG_OSD_OPREPLY:
  1198. handle_reply(osdc, msg, con);
  1199. break;
  1200. default:
  1201. pr_err("received unknown message type %d %s\n", type,
  1202. ceph_msg_type_name(type));
  1203. }
  1204. ceph_msg_put(msg);
  1205. }
  1206. /*
  1207. * lookup and return message for incoming reply
  1208. */
  1209. static struct ceph_msg *get_reply(struct ceph_connection *con,
  1210. struct ceph_msg_header *hdr,
  1211. int *skip)
  1212. {
  1213. struct ceph_osd *osd = con->private;
  1214. struct ceph_osd_client *osdc = osd->o_osdc;
  1215. struct ceph_msg *m;
  1216. struct ceph_osd_request *req;
  1217. int front = le32_to_cpu(hdr->front_len);
  1218. int data_len = le32_to_cpu(hdr->data_len);
  1219. u64 tid;
  1220. int err;
  1221. tid = le64_to_cpu(hdr->tid);
  1222. mutex_lock(&osdc->request_mutex);
  1223. req = __lookup_request(osdc, tid);
  1224. if (!req) {
  1225. *skip = 1;
  1226. m = NULL;
  1227. pr_info("alloc_msg unknown tid %llu from osd%d\n", tid,
  1228. osd->o_osd);
  1229. goto out;
  1230. }
  1231. m = __get_next_reply(con, req, front);
  1232. if (!m || IS_ERR(m)) {
  1233. *skip = 1;
  1234. goto out;
  1235. }
  1236. if (data_len > 0) {
  1237. err = __prepare_pages(con, hdr, req, tid, m);
  1238. if (err < 0) {
  1239. *skip = 1;
  1240. ceph_msg_put(m);
  1241. m = ERR_PTR(err);
  1242. }
  1243. }
  1244. *skip = 0;
  1245. out:
  1246. mutex_unlock(&osdc->request_mutex);
  1247. return m;
  1248. }
  1249. static struct ceph_msg *alloc_msg(struct ceph_connection *con,
  1250. struct ceph_msg_header *hdr,
  1251. int *skip)
  1252. {
  1253. struct ceph_osd *osd = con->private;
  1254. int type = le16_to_cpu(hdr->type);
  1255. int front = le32_to_cpu(hdr->front_len);
  1256. switch (type) {
  1257. case CEPH_MSG_OSD_MAP:
  1258. return ceph_msg_new(type, front, 0, 0, NULL);
  1259. case CEPH_MSG_OSD_OPREPLY:
  1260. return get_reply(con, hdr, skip);
  1261. default:
  1262. pr_info("alloc_msg unexpected msg type %d from osd%d\n", type,
  1263. osd->o_osd);
  1264. *skip = 1;
  1265. return NULL;
  1266. }
  1267. }
  1268. /*
  1269. * Wrappers to refcount containing ceph_osd struct
  1270. */
  1271. static struct ceph_connection *get_osd_con(struct ceph_connection *con)
  1272. {
  1273. struct ceph_osd *osd = con->private;
  1274. if (get_osd(osd))
  1275. return con;
  1276. return NULL;
  1277. }
  1278. static void put_osd_con(struct ceph_connection *con)
  1279. {
  1280. struct ceph_osd *osd = con->private;
  1281. put_osd(osd);
  1282. }
  1283. /*
  1284. * authentication
  1285. */
  1286. static int get_authorizer(struct ceph_connection *con,
  1287. void **buf, int *len, int *proto,
  1288. void **reply_buf, int *reply_len, int force_new)
  1289. {
  1290. struct ceph_osd *o = con->private;
  1291. struct ceph_osd_client *osdc = o->o_osdc;
  1292. struct ceph_auth_client *ac = osdc->client->monc.auth;
  1293. int ret = 0;
  1294. if (force_new && o->o_authorizer) {
  1295. ac->ops->destroy_authorizer(ac, o->o_authorizer);
  1296. o->o_authorizer = NULL;
  1297. }
  1298. if (o->o_authorizer == NULL) {
  1299. ret = ac->ops->create_authorizer(
  1300. ac, CEPH_ENTITY_TYPE_OSD,
  1301. &o->o_authorizer,
  1302. &o->o_authorizer_buf,
  1303. &o->o_authorizer_buf_len,
  1304. &o->o_authorizer_reply_buf,
  1305. &o->o_authorizer_reply_buf_len);
  1306. if (ret)
  1307. return ret;
  1308. }
  1309. *proto = ac->protocol;
  1310. *buf = o->o_authorizer_buf;
  1311. *len = o->o_authorizer_buf_len;
  1312. *reply_buf = o->o_authorizer_reply_buf;
  1313. *reply_len = o->o_authorizer_reply_buf_len;
  1314. return 0;
  1315. }
  1316. static int verify_authorizer_reply(struct ceph_connection *con, int len)
  1317. {
  1318. struct ceph_osd *o = con->private;
  1319. struct ceph_osd_client *osdc = o->o_osdc;
  1320. struct ceph_auth_client *ac = osdc->client->monc.auth;
  1321. return ac->ops->verify_authorizer_reply(ac, o->o_authorizer, len);
  1322. }
  1323. static int invalidate_authorizer(struct ceph_connection *con)
  1324. {
  1325. struct ceph_osd *o = con->private;
  1326. struct ceph_osd_client *osdc = o->o_osdc;
  1327. struct ceph_auth_client *ac = osdc->client->monc.auth;
  1328. if (ac->ops->invalidate_authorizer)
  1329. ac->ops->invalidate_authorizer(ac, CEPH_ENTITY_TYPE_OSD);
  1330. return ceph_monc_validate_auth(&osdc->client->monc);
  1331. }
  1332. const static struct ceph_connection_operations osd_con_ops = {
  1333. .get = get_osd_con,
  1334. .put = put_osd_con,
  1335. .dispatch = dispatch,
  1336. .get_authorizer = get_authorizer,
  1337. .verify_authorizer_reply = verify_authorizer_reply,
  1338. .invalidate_authorizer = invalidate_authorizer,
  1339. .alloc_msg = alloc_msg,
  1340. .fault = osd_reset,
  1341. };