osd_client.c 39 KB

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