mds_client.c 78 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875287628772878287928802881288228832884288528862887288828892890289128922893289428952896289728982899290029012902290329042905290629072908290929102911291229132914291529162917291829192920292129222923292429252926292729282929293029312932293329342935293629372938293929402941294229432944294529462947294829492950295129522953295429552956295729582959296029612962296329642965296629672968296929702971297229732974297529762977297829792980298129822983298429852986298729882989299029912992299329942995299629972998299930003001300230033004300530063007300830093010301130123013301430153016301730183019302030213022302330243025302630273028302930303031303230333034303530363037303830393040304130423043304430453046304730483049305030513052305330543055305630573058305930603061306230633064306530663067306830693070307130723073307430753076307730783079308030813082308330843085308630873088308930903091309230933094309530963097309830993100310131023103310431053106310731083109311031113112311331143115311631173118311931203121312231233124312531263127312831293130313131323133313431353136
  1. #include "ceph_debug.h"
  2. #include <linux/wait.h>
  3. #include <linux/slab.h>
  4. #include <linux/sched.h>
  5. #include "mds_client.h"
  6. #include "mon_client.h"
  7. #include "super.h"
  8. #include "messenger.h"
  9. #include "decode.h"
  10. #include "auth.h"
  11. #include "pagelist.h"
  12. /*
  13. * A cluster of MDS (metadata server) daemons is responsible for
  14. * managing the file system namespace (the directory hierarchy and
  15. * inodes) and for coordinating shared access to storage. Metadata is
  16. * partitioning hierarchically across a number of servers, and that
  17. * partition varies over time as the cluster adjusts the distribution
  18. * in order to balance load.
  19. *
  20. * The MDS client is primarily responsible to managing synchronous
  21. * metadata requests for operations like open, unlink, and so forth.
  22. * If there is a MDS failure, we find out about it when we (possibly
  23. * request and) receive a new MDS map, and can resubmit affected
  24. * requests.
  25. *
  26. * For the most part, though, we take advantage of a lossless
  27. * communications channel to the MDS, and do not need to worry about
  28. * timing out or resubmitting requests.
  29. *
  30. * We maintain a stateful "session" with each MDS we interact with.
  31. * Within each session, we sent periodic heartbeat messages to ensure
  32. * any capabilities or leases we have been issues remain valid. If
  33. * the session times out and goes stale, our leases and capabilities
  34. * are no longer valid.
  35. */
  36. static void __wake_requests(struct ceph_mds_client *mdsc,
  37. struct list_head *head);
  38. static const struct ceph_connection_operations mds_con_ops;
  39. /*
  40. * mds reply parsing
  41. */
  42. /*
  43. * parse individual inode info
  44. */
  45. static int parse_reply_info_in(void **p, void *end,
  46. struct ceph_mds_reply_info_in *info)
  47. {
  48. int err = -EIO;
  49. info->in = *p;
  50. *p += sizeof(struct ceph_mds_reply_inode) +
  51. sizeof(*info->in->fragtree.splits) *
  52. le32_to_cpu(info->in->fragtree.nsplits);
  53. ceph_decode_32_safe(p, end, info->symlink_len, bad);
  54. ceph_decode_need(p, end, info->symlink_len, bad);
  55. info->symlink = *p;
  56. *p += info->symlink_len;
  57. ceph_decode_32_safe(p, end, info->xattr_len, bad);
  58. ceph_decode_need(p, end, info->xattr_len, bad);
  59. info->xattr_data = *p;
  60. *p += info->xattr_len;
  61. return 0;
  62. bad:
  63. return err;
  64. }
  65. /*
  66. * parse a normal reply, which may contain a (dir+)dentry and/or a
  67. * target inode.
  68. */
  69. static int parse_reply_info_trace(void **p, void *end,
  70. struct ceph_mds_reply_info_parsed *info)
  71. {
  72. int err;
  73. if (info->head->is_dentry) {
  74. err = parse_reply_info_in(p, end, &info->diri);
  75. if (err < 0)
  76. goto out_bad;
  77. if (unlikely(*p + sizeof(*info->dirfrag) > end))
  78. goto bad;
  79. info->dirfrag = *p;
  80. *p += sizeof(*info->dirfrag) +
  81. sizeof(u32)*le32_to_cpu(info->dirfrag->ndist);
  82. if (unlikely(*p > end))
  83. goto bad;
  84. ceph_decode_32_safe(p, end, info->dname_len, bad);
  85. ceph_decode_need(p, end, info->dname_len, bad);
  86. info->dname = *p;
  87. *p += info->dname_len;
  88. info->dlease = *p;
  89. *p += sizeof(*info->dlease);
  90. }
  91. if (info->head->is_target) {
  92. err = parse_reply_info_in(p, end, &info->targeti);
  93. if (err < 0)
  94. goto out_bad;
  95. }
  96. if (unlikely(*p != end))
  97. goto bad;
  98. return 0;
  99. bad:
  100. err = -EIO;
  101. out_bad:
  102. pr_err("problem parsing mds trace %d\n", err);
  103. return err;
  104. }
  105. /*
  106. * parse readdir results
  107. */
  108. static int parse_reply_info_dir(void **p, void *end,
  109. struct ceph_mds_reply_info_parsed *info)
  110. {
  111. u32 num, i = 0;
  112. int err;
  113. info->dir_dir = *p;
  114. if (*p + sizeof(*info->dir_dir) > end)
  115. goto bad;
  116. *p += sizeof(*info->dir_dir) +
  117. sizeof(u32)*le32_to_cpu(info->dir_dir->ndist);
  118. if (*p > end)
  119. goto bad;
  120. ceph_decode_need(p, end, sizeof(num) + 2, bad);
  121. num = ceph_decode_32(p);
  122. info->dir_end = ceph_decode_8(p);
  123. info->dir_complete = ceph_decode_8(p);
  124. if (num == 0)
  125. goto done;
  126. /* alloc large array */
  127. info->dir_nr = num;
  128. info->dir_in = kcalloc(num, sizeof(*info->dir_in) +
  129. sizeof(*info->dir_dname) +
  130. sizeof(*info->dir_dname_len) +
  131. sizeof(*info->dir_dlease),
  132. GFP_NOFS);
  133. if (info->dir_in == NULL) {
  134. err = -ENOMEM;
  135. goto out_bad;
  136. }
  137. info->dir_dname = (void *)(info->dir_in + num);
  138. info->dir_dname_len = (void *)(info->dir_dname + num);
  139. info->dir_dlease = (void *)(info->dir_dname_len + num);
  140. while (num) {
  141. /* dentry */
  142. ceph_decode_need(p, end, sizeof(u32)*2, bad);
  143. info->dir_dname_len[i] = ceph_decode_32(p);
  144. ceph_decode_need(p, end, info->dir_dname_len[i], bad);
  145. info->dir_dname[i] = *p;
  146. *p += info->dir_dname_len[i];
  147. dout("parsed dir dname '%.*s'\n", info->dir_dname_len[i],
  148. info->dir_dname[i]);
  149. info->dir_dlease[i] = *p;
  150. *p += sizeof(struct ceph_mds_reply_lease);
  151. /* inode */
  152. err = parse_reply_info_in(p, end, &info->dir_in[i]);
  153. if (err < 0)
  154. goto out_bad;
  155. i++;
  156. num--;
  157. }
  158. done:
  159. if (*p != end)
  160. goto bad;
  161. return 0;
  162. bad:
  163. err = -EIO;
  164. out_bad:
  165. pr_err("problem parsing dir contents %d\n", err);
  166. return err;
  167. }
  168. /*
  169. * parse entire mds reply
  170. */
  171. static int parse_reply_info(struct ceph_msg *msg,
  172. struct ceph_mds_reply_info_parsed *info)
  173. {
  174. void *p, *end;
  175. u32 len;
  176. int err;
  177. info->head = msg->front.iov_base;
  178. p = msg->front.iov_base + sizeof(struct ceph_mds_reply_head);
  179. end = p + msg->front.iov_len - sizeof(struct ceph_mds_reply_head);
  180. /* trace */
  181. ceph_decode_32_safe(&p, end, len, bad);
  182. if (len > 0) {
  183. err = parse_reply_info_trace(&p, p+len, info);
  184. if (err < 0)
  185. goto out_bad;
  186. }
  187. /* dir content */
  188. ceph_decode_32_safe(&p, end, len, bad);
  189. if (len > 0) {
  190. err = parse_reply_info_dir(&p, p+len, info);
  191. if (err < 0)
  192. goto out_bad;
  193. }
  194. /* snap blob */
  195. ceph_decode_32_safe(&p, end, len, bad);
  196. info->snapblob_len = len;
  197. info->snapblob = p;
  198. p += len;
  199. if (p != end)
  200. goto bad;
  201. return 0;
  202. bad:
  203. err = -EIO;
  204. out_bad:
  205. pr_err("mds parse_reply err %d\n", err);
  206. return err;
  207. }
  208. static void destroy_reply_info(struct ceph_mds_reply_info_parsed *info)
  209. {
  210. kfree(info->dir_in);
  211. }
  212. /*
  213. * sessions
  214. */
  215. static const char *session_state_name(int s)
  216. {
  217. switch (s) {
  218. case CEPH_MDS_SESSION_NEW: return "new";
  219. case CEPH_MDS_SESSION_OPENING: return "opening";
  220. case CEPH_MDS_SESSION_OPEN: return "open";
  221. case CEPH_MDS_SESSION_HUNG: return "hung";
  222. case CEPH_MDS_SESSION_CLOSING: return "closing";
  223. case CEPH_MDS_SESSION_RESTARTING: return "restarting";
  224. case CEPH_MDS_SESSION_RECONNECTING: return "reconnecting";
  225. default: return "???";
  226. }
  227. }
  228. static struct ceph_mds_session *get_session(struct ceph_mds_session *s)
  229. {
  230. if (atomic_inc_not_zero(&s->s_ref)) {
  231. dout("mdsc get_session %p %d -> %d\n", s,
  232. atomic_read(&s->s_ref)-1, atomic_read(&s->s_ref));
  233. return s;
  234. } else {
  235. dout("mdsc get_session %p 0 -- FAIL", s);
  236. return NULL;
  237. }
  238. }
  239. void ceph_put_mds_session(struct ceph_mds_session *s)
  240. {
  241. dout("mdsc put_session %p %d -> %d\n", s,
  242. atomic_read(&s->s_ref), atomic_read(&s->s_ref)-1);
  243. if (atomic_dec_and_test(&s->s_ref)) {
  244. if (s->s_authorizer)
  245. s->s_mdsc->client->monc.auth->ops->destroy_authorizer(
  246. s->s_mdsc->client->monc.auth, s->s_authorizer);
  247. kfree(s);
  248. }
  249. }
  250. /*
  251. * called under mdsc->mutex
  252. */
  253. struct ceph_mds_session *__ceph_lookup_mds_session(struct ceph_mds_client *mdsc,
  254. int mds)
  255. {
  256. struct ceph_mds_session *session;
  257. if (mds >= mdsc->max_sessions || mdsc->sessions[mds] == NULL)
  258. return NULL;
  259. session = mdsc->sessions[mds];
  260. dout("lookup_mds_session %p %d\n", session,
  261. atomic_read(&session->s_ref));
  262. get_session(session);
  263. return session;
  264. }
  265. static bool __have_session(struct ceph_mds_client *mdsc, int mds)
  266. {
  267. if (mds >= mdsc->max_sessions)
  268. return false;
  269. return mdsc->sessions[mds];
  270. }
  271. static int __verify_registered_session(struct ceph_mds_client *mdsc,
  272. struct ceph_mds_session *s)
  273. {
  274. if (s->s_mds >= mdsc->max_sessions ||
  275. mdsc->sessions[s->s_mds] != s)
  276. return -ENOENT;
  277. return 0;
  278. }
  279. /*
  280. * create+register a new session for given mds.
  281. * called under mdsc->mutex.
  282. */
  283. static struct ceph_mds_session *register_session(struct ceph_mds_client *mdsc,
  284. int mds)
  285. {
  286. struct ceph_mds_session *s;
  287. s = kzalloc(sizeof(*s), GFP_NOFS);
  288. if (!s)
  289. return ERR_PTR(-ENOMEM);
  290. s->s_mdsc = mdsc;
  291. s->s_mds = mds;
  292. s->s_state = CEPH_MDS_SESSION_NEW;
  293. s->s_ttl = 0;
  294. s->s_seq = 0;
  295. mutex_init(&s->s_mutex);
  296. ceph_con_init(mdsc->client->msgr, &s->s_con);
  297. s->s_con.private = s;
  298. s->s_con.ops = &mds_con_ops;
  299. s->s_con.peer_name.type = CEPH_ENTITY_TYPE_MDS;
  300. s->s_con.peer_name.num = cpu_to_le64(mds);
  301. spin_lock_init(&s->s_cap_lock);
  302. s->s_cap_gen = 0;
  303. s->s_cap_ttl = 0;
  304. s->s_renew_requested = 0;
  305. s->s_renew_seq = 0;
  306. INIT_LIST_HEAD(&s->s_caps);
  307. s->s_nr_caps = 0;
  308. s->s_trim_caps = 0;
  309. atomic_set(&s->s_ref, 1);
  310. INIT_LIST_HEAD(&s->s_waiting);
  311. INIT_LIST_HEAD(&s->s_unsafe);
  312. s->s_num_cap_releases = 0;
  313. s->s_cap_iterator = NULL;
  314. INIT_LIST_HEAD(&s->s_cap_releases);
  315. INIT_LIST_HEAD(&s->s_cap_releases_done);
  316. INIT_LIST_HEAD(&s->s_cap_flushing);
  317. INIT_LIST_HEAD(&s->s_cap_snaps_flushing);
  318. dout("register_session mds%d\n", mds);
  319. if (mds >= mdsc->max_sessions) {
  320. int newmax = 1 << get_count_order(mds+1);
  321. struct ceph_mds_session **sa;
  322. dout("register_session realloc to %d\n", newmax);
  323. sa = kcalloc(newmax, sizeof(void *), GFP_NOFS);
  324. if (sa == NULL)
  325. goto fail_realloc;
  326. if (mdsc->sessions) {
  327. memcpy(sa, mdsc->sessions,
  328. mdsc->max_sessions * sizeof(void *));
  329. kfree(mdsc->sessions);
  330. }
  331. mdsc->sessions = sa;
  332. mdsc->max_sessions = newmax;
  333. }
  334. mdsc->sessions[mds] = s;
  335. atomic_inc(&s->s_ref); /* one ref to sessions[], one to caller */
  336. ceph_con_open(&s->s_con, ceph_mdsmap_get_addr(mdsc->mdsmap, mds));
  337. return s;
  338. fail_realloc:
  339. kfree(s);
  340. return ERR_PTR(-ENOMEM);
  341. }
  342. /*
  343. * called under mdsc->mutex
  344. */
  345. static void __unregister_session(struct ceph_mds_client *mdsc,
  346. struct ceph_mds_session *s)
  347. {
  348. dout("__unregister_session mds%d %p\n", s->s_mds, s);
  349. BUG_ON(mdsc->sessions[s->s_mds] != s);
  350. mdsc->sessions[s->s_mds] = NULL;
  351. ceph_con_close(&s->s_con);
  352. ceph_put_mds_session(s);
  353. }
  354. /*
  355. * drop session refs in request.
  356. *
  357. * should be last request ref, or hold mdsc->mutex
  358. */
  359. static void put_request_session(struct ceph_mds_request *req)
  360. {
  361. if (req->r_session) {
  362. ceph_put_mds_session(req->r_session);
  363. req->r_session = NULL;
  364. }
  365. }
  366. void ceph_mdsc_release_request(struct kref *kref)
  367. {
  368. struct ceph_mds_request *req = container_of(kref,
  369. struct ceph_mds_request,
  370. r_kref);
  371. if (req->r_request)
  372. ceph_msg_put(req->r_request);
  373. if (req->r_reply) {
  374. ceph_msg_put(req->r_reply);
  375. destroy_reply_info(&req->r_reply_info);
  376. }
  377. if (req->r_inode) {
  378. ceph_put_cap_refs(ceph_inode(req->r_inode),
  379. CEPH_CAP_PIN);
  380. iput(req->r_inode);
  381. }
  382. if (req->r_locked_dir)
  383. ceph_put_cap_refs(ceph_inode(req->r_locked_dir),
  384. CEPH_CAP_PIN);
  385. if (req->r_target_inode)
  386. iput(req->r_target_inode);
  387. if (req->r_dentry)
  388. dput(req->r_dentry);
  389. if (req->r_old_dentry) {
  390. ceph_put_cap_refs(
  391. ceph_inode(req->r_old_dentry->d_parent->d_inode),
  392. CEPH_CAP_PIN);
  393. dput(req->r_old_dentry);
  394. }
  395. kfree(req->r_path1);
  396. kfree(req->r_path2);
  397. put_request_session(req);
  398. ceph_unreserve_caps(&req->r_caps_reservation);
  399. kfree(req);
  400. }
  401. /*
  402. * lookup session, bump ref if found.
  403. *
  404. * called under mdsc->mutex.
  405. */
  406. static struct ceph_mds_request *__lookup_request(struct ceph_mds_client *mdsc,
  407. u64 tid)
  408. {
  409. struct ceph_mds_request *req;
  410. struct rb_node *n = mdsc->request_tree.rb_node;
  411. while (n) {
  412. req = rb_entry(n, struct ceph_mds_request, r_node);
  413. if (tid < req->r_tid)
  414. n = n->rb_left;
  415. else if (tid > req->r_tid)
  416. n = n->rb_right;
  417. else {
  418. ceph_mdsc_get_request(req);
  419. return req;
  420. }
  421. }
  422. return NULL;
  423. }
  424. static void __insert_request(struct ceph_mds_client *mdsc,
  425. struct ceph_mds_request *new)
  426. {
  427. struct rb_node **p = &mdsc->request_tree.rb_node;
  428. struct rb_node *parent = NULL;
  429. struct ceph_mds_request *req = NULL;
  430. while (*p) {
  431. parent = *p;
  432. req = rb_entry(parent, struct ceph_mds_request, r_node);
  433. if (new->r_tid < req->r_tid)
  434. p = &(*p)->rb_left;
  435. else if (new->r_tid > req->r_tid)
  436. p = &(*p)->rb_right;
  437. else
  438. BUG();
  439. }
  440. rb_link_node(&new->r_node, parent, p);
  441. rb_insert_color(&new->r_node, &mdsc->request_tree);
  442. }
  443. /*
  444. * Register an in-flight request, and assign a tid. Link to directory
  445. * are modifying (if any).
  446. *
  447. * Called under mdsc->mutex.
  448. */
  449. static void __register_request(struct ceph_mds_client *mdsc,
  450. struct ceph_mds_request *req,
  451. struct inode *dir)
  452. {
  453. req->r_tid = ++mdsc->last_tid;
  454. if (req->r_num_caps)
  455. ceph_reserve_caps(&req->r_caps_reservation, req->r_num_caps);
  456. dout("__register_request %p tid %lld\n", req, req->r_tid);
  457. ceph_mdsc_get_request(req);
  458. __insert_request(mdsc, req);
  459. if (dir) {
  460. struct ceph_inode_info *ci = ceph_inode(dir);
  461. spin_lock(&ci->i_unsafe_lock);
  462. req->r_unsafe_dir = dir;
  463. list_add_tail(&req->r_unsafe_dir_item, &ci->i_unsafe_dirops);
  464. spin_unlock(&ci->i_unsafe_lock);
  465. }
  466. }
  467. static void __unregister_request(struct ceph_mds_client *mdsc,
  468. struct ceph_mds_request *req)
  469. {
  470. dout("__unregister_request %p tid %lld\n", req, req->r_tid);
  471. rb_erase(&req->r_node, &mdsc->request_tree);
  472. RB_CLEAR_NODE(&req->r_node);
  473. if (req->r_unsafe_dir) {
  474. struct ceph_inode_info *ci = ceph_inode(req->r_unsafe_dir);
  475. spin_lock(&ci->i_unsafe_lock);
  476. list_del_init(&req->r_unsafe_dir_item);
  477. spin_unlock(&ci->i_unsafe_lock);
  478. }
  479. ceph_mdsc_put_request(req);
  480. }
  481. /*
  482. * Choose mds to send request to next. If there is a hint set in the
  483. * request (e.g., due to a prior forward hint from the mds), use that.
  484. * Otherwise, consult frag tree and/or caps to identify the
  485. * appropriate mds. If all else fails, choose randomly.
  486. *
  487. * Called under mdsc->mutex.
  488. */
  489. static int __choose_mds(struct ceph_mds_client *mdsc,
  490. struct ceph_mds_request *req)
  491. {
  492. struct inode *inode;
  493. struct ceph_inode_info *ci;
  494. struct ceph_cap *cap;
  495. int mode = req->r_direct_mode;
  496. int mds = -1;
  497. u32 hash = req->r_direct_hash;
  498. bool is_hash = req->r_direct_is_hash;
  499. /*
  500. * is there a specific mds we should try? ignore hint if we have
  501. * no session and the mds is not up (active or recovering).
  502. */
  503. if (req->r_resend_mds >= 0 &&
  504. (__have_session(mdsc, req->r_resend_mds) ||
  505. ceph_mdsmap_get_state(mdsc->mdsmap, req->r_resend_mds) > 0)) {
  506. dout("choose_mds using resend_mds mds%d\n",
  507. req->r_resend_mds);
  508. return req->r_resend_mds;
  509. }
  510. if (mode == USE_RANDOM_MDS)
  511. goto random;
  512. inode = NULL;
  513. if (req->r_inode) {
  514. inode = req->r_inode;
  515. } else if (req->r_dentry) {
  516. if (req->r_dentry->d_inode) {
  517. inode = req->r_dentry->d_inode;
  518. } else {
  519. inode = req->r_dentry->d_parent->d_inode;
  520. hash = req->r_dentry->d_name.hash;
  521. is_hash = true;
  522. }
  523. }
  524. dout("__choose_mds %p is_hash=%d (%d) mode %d\n", inode, (int)is_hash,
  525. (int)hash, mode);
  526. if (!inode)
  527. goto random;
  528. ci = ceph_inode(inode);
  529. if (is_hash && S_ISDIR(inode->i_mode)) {
  530. struct ceph_inode_frag frag;
  531. int found;
  532. ceph_choose_frag(ci, hash, &frag, &found);
  533. if (found) {
  534. if (mode == USE_ANY_MDS && frag.ndist > 0) {
  535. u8 r;
  536. /* choose a random replica */
  537. get_random_bytes(&r, 1);
  538. r %= frag.ndist;
  539. mds = frag.dist[r];
  540. dout("choose_mds %p %llx.%llx "
  541. "frag %u mds%d (%d/%d)\n",
  542. inode, ceph_vinop(inode),
  543. frag.frag, frag.mds,
  544. (int)r, frag.ndist);
  545. return mds;
  546. }
  547. /* since this file/dir wasn't known to be
  548. * replicated, then we want to look for the
  549. * authoritative mds. */
  550. mode = USE_AUTH_MDS;
  551. if (frag.mds >= 0) {
  552. /* choose auth mds */
  553. mds = frag.mds;
  554. dout("choose_mds %p %llx.%llx "
  555. "frag %u mds%d (auth)\n",
  556. inode, ceph_vinop(inode), frag.frag, mds);
  557. return mds;
  558. }
  559. }
  560. }
  561. spin_lock(&inode->i_lock);
  562. cap = NULL;
  563. if (mode == USE_AUTH_MDS)
  564. cap = ci->i_auth_cap;
  565. if (!cap && !RB_EMPTY_ROOT(&ci->i_caps))
  566. cap = rb_entry(rb_first(&ci->i_caps), struct ceph_cap, ci_node);
  567. if (!cap) {
  568. spin_unlock(&inode->i_lock);
  569. goto random;
  570. }
  571. mds = cap->session->s_mds;
  572. dout("choose_mds %p %llx.%llx mds%d (%scap %p)\n",
  573. inode, ceph_vinop(inode), mds,
  574. cap == ci->i_auth_cap ? "auth " : "", cap);
  575. spin_unlock(&inode->i_lock);
  576. return mds;
  577. random:
  578. mds = ceph_mdsmap_get_random_mds(mdsc->mdsmap);
  579. dout("choose_mds chose random mds%d\n", mds);
  580. return mds;
  581. }
  582. /*
  583. * session messages
  584. */
  585. static struct ceph_msg *create_session_msg(u32 op, u64 seq)
  586. {
  587. struct ceph_msg *msg;
  588. struct ceph_mds_session_head *h;
  589. msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h), GFP_NOFS);
  590. if (!msg) {
  591. pr_err("create_session_msg ENOMEM creating msg\n");
  592. return NULL;
  593. }
  594. h = msg->front.iov_base;
  595. h->op = cpu_to_le32(op);
  596. h->seq = cpu_to_le64(seq);
  597. return msg;
  598. }
  599. /*
  600. * send session open request.
  601. *
  602. * called under mdsc->mutex
  603. */
  604. static int __open_session(struct ceph_mds_client *mdsc,
  605. struct ceph_mds_session *session)
  606. {
  607. struct ceph_msg *msg;
  608. int mstate;
  609. int mds = session->s_mds;
  610. /* wait for mds to go active? */
  611. mstate = ceph_mdsmap_get_state(mdsc->mdsmap, mds);
  612. dout("open_session to mds%d (%s)\n", mds,
  613. ceph_mds_state_name(mstate));
  614. session->s_state = CEPH_MDS_SESSION_OPENING;
  615. session->s_renew_requested = jiffies;
  616. /* send connect message */
  617. msg = create_session_msg(CEPH_SESSION_REQUEST_OPEN, session->s_seq);
  618. if (!msg)
  619. return -ENOMEM;
  620. ceph_con_send(&session->s_con, msg);
  621. return 0;
  622. }
  623. /*
  624. * session caps
  625. */
  626. /*
  627. * Free preallocated cap messages assigned to this session
  628. */
  629. static void cleanup_cap_releases(struct ceph_mds_session *session)
  630. {
  631. struct ceph_msg *msg;
  632. spin_lock(&session->s_cap_lock);
  633. while (!list_empty(&session->s_cap_releases)) {
  634. msg = list_first_entry(&session->s_cap_releases,
  635. struct ceph_msg, list_head);
  636. list_del_init(&msg->list_head);
  637. ceph_msg_put(msg);
  638. }
  639. while (!list_empty(&session->s_cap_releases_done)) {
  640. msg = list_first_entry(&session->s_cap_releases_done,
  641. struct ceph_msg, list_head);
  642. list_del_init(&msg->list_head);
  643. ceph_msg_put(msg);
  644. }
  645. spin_unlock(&session->s_cap_lock);
  646. }
  647. /*
  648. * Helper to safely iterate over all caps associated with a session, with
  649. * special care taken to handle a racing __ceph_remove_cap().
  650. *
  651. * Caller must hold session s_mutex.
  652. */
  653. static int iterate_session_caps(struct ceph_mds_session *session,
  654. int (*cb)(struct inode *, struct ceph_cap *,
  655. void *), void *arg)
  656. {
  657. struct list_head *p;
  658. struct ceph_cap *cap;
  659. struct inode *inode, *last_inode = NULL;
  660. struct ceph_cap *old_cap = NULL;
  661. int ret;
  662. dout("iterate_session_caps %p mds%d\n", session, session->s_mds);
  663. spin_lock(&session->s_cap_lock);
  664. p = session->s_caps.next;
  665. while (p != &session->s_caps) {
  666. cap = list_entry(p, struct ceph_cap, session_caps);
  667. inode = igrab(&cap->ci->vfs_inode);
  668. if (!inode) {
  669. p = p->next;
  670. continue;
  671. }
  672. session->s_cap_iterator = cap;
  673. spin_unlock(&session->s_cap_lock);
  674. if (last_inode) {
  675. iput(last_inode);
  676. last_inode = NULL;
  677. }
  678. if (old_cap) {
  679. ceph_put_cap(old_cap);
  680. old_cap = NULL;
  681. }
  682. ret = cb(inode, cap, arg);
  683. last_inode = inode;
  684. spin_lock(&session->s_cap_lock);
  685. p = p->next;
  686. if (cap->ci == NULL) {
  687. dout("iterate_session_caps finishing cap %p removal\n",
  688. cap);
  689. BUG_ON(cap->session != session);
  690. list_del_init(&cap->session_caps);
  691. session->s_nr_caps--;
  692. cap->session = NULL;
  693. old_cap = cap; /* put_cap it w/o locks held */
  694. }
  695. if (ret < 0)
  696. goto out;
  697. }
  698. ret = 0;
  699. out:
  700. session->s_cap_iterator = NULL;
  701. spin_unlock(&session->s_cap_lock);
  702. if (last_inode)
  703. iput(last_inode);
  704. if (old_cap)
  705. ceph_put_cap(old_cap);
  706. return ret;
  707. }
  708. static int remove_session_caps_cb(struct inode *inode, struct ceph_cap *cap,
  709. void *arg)
  710. {
  711. struct ceph_inode_info *ci = ceph_inode(inode);
  712. int drop = 0;
  713. dout("removing cap %p, ci is %p, inode is %p\n",
  714. cap, ci, &ci->vfs_inode);
  715. spin_lock(&inode->i_lock);
  716. __ceph_remove_cap(cap);
  717. if (!__ceph_is_any_real_caps(ci)) {
  718. struct ceph_mds_client *mdsc =
  719. &ceph_sb_to_client(inode->i_sb)->mdsc;
  720. spin_lock(&mdsc->cap_dirty_lock);
  721. if (!list_empty(&ci->i_dirty_item)) {
  722. pr_info(" dropping dirty %s state for %p %lld\n",
  723. ceph_cap_string(ci->i_dirty_caps),
  724. inode, ceph_ino(inode));
  725. ci->i_dirty_caps = 0;
  726. list_del_init(&ci->i_dirty_item);
  727. drop = 1;
  728. }
  729. if (!list_empty(&ci->i_flushing_item)) {
  730. pr_info(" dropping dirty+flushing %s state for %p %lld\n",
  731. ceph_cap_string(ci->i_flushing_caps),
  732. inode, ceph_ino(inode));
  733. ci->i_flushing_caps = 0;
  734. list_del_init(&ci->i_flushing_item);
  735. mdsc->num_cap_flushing--;
  736. drop = 1;
  737. }
  738. if (drop && ci->i_wrbuffer_ref) {
  739. pr_info(" dropping dirty data for %p %lld\n",
  740. inode, ceph_ino(inode));
  741. ci->i_wrbuffer_ref = 0;
  742. ci->i_wrbuffer_ref_head = 0;
  743. drop++;
  744. }
  745. spin_unlock(&mdsc->cap_dirty_lock);
  746. }
  747. spin_unlock(&inode->i_lock);
  748. while (drop--)
  749. iput(inode);
  750. return 0;
  751. }
  752. /*
  753. * caller must hold session s_mutex
  754. */
  755. static void remove_session_caps(struct ceph_mds_session *session)
  756. {
  757. dout("remove_session_caps on %p\n", session);
  758. iterate_session_caps(session, remove_session_caps_cb, NULL);
  759. BUG_ON(session->s_nr_caps > 0);
  760. BUG_ON(!list_empty(&session->s_cap_flushing));
  761. cleanup_cap_releases(session);
  762. }
  763. /*
  764. * wake up any threads waiting on this session's caps. if the cap is
  765. * old (didn't get renewed on the client reconnect), remove it now.
  766. *
  767. * caller must hold s_mutex.
  768. */
  769. static int wake_up_session_cb(struct inode *inode, struct ceph_cap *cap,
  770. void *arg)
  771. {
  772. struct ceph_inode_info *ci = ceph_inode(inode);
  773. wake_up(&ci->i_cap_wq);
  774. if (arg) {
  775. spin_lock(&inode->i_lock);
  776. ci->i_wanted_max_size = 0;
  777. ci->i_requested_max_size = 0;
  778. spin_unlock(&inode->i_lock);
  779. }
  780. return 0;
  781. }
  782. static void wake_up_session_caps(struct ceph_mds_session *session,
  783. int reconnect)
  784. {
  785. dout("wake_up_session_caps %p mds%d\n", session, session->s_mds);
  786. iterate_session_caps(session, wake_up_session_cb,
  787. (void *)(unsigned long)reconnect);
  788. }
  789. /*
  790. * Send periodic message to MDS renewing all currently held caps. The
  791. * ack will reset the expiration for all caps from this session.
  792. *
  793. * caller holds s_mutex
  794. */
  795. static int send_renew_caps(struct ceph_mds_client *mdsc,
  796. struct ceph_mds_session *session)
  797. {
  798. struct ceph_msg *msg;
  799. int state;
  800. if (time_after_eq(jiffies, session->s_cap_ttl) &&
  801. time_after_eq(session->s_cap_ttl, session->s_renew_requested))
  802. pr_info("mds%d caps stale\n", session->s_mds);
  803. session->s_renew_requested = jiffies;
  804. /* do not try to renew caps until a recovering mds has reconnected
  805. * with its clients. */
  806. state = ceph_mdsmap_get_state(mdsc->mdsmap, session->s_mds);
  807. if (state < CEPH_MDS_STATE_RECONNECT) {
  808. dout("send_renew_caps ignoring mds%d (%s)\n",
  809. session->s_mds, ceph_mds_state_name(state));
  810. return 0;
  811. }
  812. dout("send_renew_caps to mds%d (%s)\n", session->s_mds,
  813. ceph_mds_state_name(state));
  814. msg = create_session_msg(CEPH_SESSION_REQUEST_RENEWCAPS,
  815. ++session->s_renew_seq);
  816. if (!msg)
  817. return -ENOMEM;
  818. ceph_con_send(&session->s_con, msg);
  819. return 0;
  820. }
  821. /*
  822. * Note new cap ttl, and any transition from stale -> not stale (fresh?).
  823. *
  824. * Called under session->s_mutex
  825. */
  826. static void renewed_caps(struct ceph_mds_client *mdsc,
  827. struct ceph_mds_session *session, int is_renew)
  828. {
  829. int was_stale;
  830. int wake = 0;
  831. spin_lock(&session->s_cap_lock);
  832. was_stale = is_renew && (session->s_cap_ttl == 0 ||
  833. time_after_eq(jiffies, session->s_cap_ttl));
  834. session->s_cap_ttl = session->s_renew_requested +
  835. mdsc->mdsmap->m_session_timeout*HZ;
  836. if (was_stale) {
  837. if (time_before(jiffies, session->s_cap_ttl)) {
  838. pr_info("mds%d caps renewed\n", session->s_mds);
  839. wake = 1;
  840. } else {
  841. pr_info("mds%d caps still stale\n", session->s_mds);
  842. }
  843. }
  844. dout("renewed_caps mds%d ttl now %lu, was %s, now %s\n",
  845. session->s_mds, session->s_cap_ttl, was_stale ? "stale" : "fresh",
  846. time_before(jiffies, session->s_cap_ttl) ? "stale" : "fresh");
  847. spin_unlock(&session->s_cap_lock);
  848. if (wake)
  849. wake_up_session_caps(session, 0);
  850. }
  851. /*
  852. * send a session close request
  853. */
  854. static int request_close_session(struct ceph_mds_client *mdsc,
  855. struct ceph_mds_session *session)
  856. {
  857. struct ceph_msg *msg;
  858. dout("request_close_session mds%d state %s seq %lld\n",
  859. session->s_mds, session_state_name(session->s_state),
  860. session->s_seq);
  861. msg = create_session_msg(CEPH_SESSION_REQUEST_CLOSE, session->s_seq);
  862. if (!msg)
  863. return -ENOMEM;
  864. ceph_con_send(&session->s_con, msg);
  865. return 0;
  866. }
  867. /*
  868. * Called with s_mutex held.
  869. */
  870. static int __close_session(struct ceph_mds_client *mdsc,
  871. struct ceph_mds_session *session)
  872. {
  873. if (session->s_state >= CEPH_MDS_SESSION_CLOSING)
  874. return 0;
  875. session->s_state = CEPH_MDS_SESSION_CLOSING;
  876. return request_close_session(mdsc, session);
  877. }
  878. /*
  879. * Trim old(er) caps.
  880. *
  881. * Because we can't cache an inode without one or more caps, we do
  882. * this indirectly: if a cap is unused, we prune its aliases, at which
  883. * point the inode will hopefully get dropped to.
  884. *
  885. * Yes, this is a bit sloppy. Our only real goal here is to respond to
  886. * memory pressure from the MDS, though, so it needn't be perfect.
  887. */
  888. static int trim_caps_cb(struct inode *inode, struct ceph_cap *cap, void *arg)
  889. {
  890. struct ceph_mds_session *session = arg;
  891. struct ceph_inode_info *ci = ceph_inode(inode);
  892. int used, oissued, mine;
  893. if (session->s_trim_caps <= 0)
  894. return -1;
  895. spin_lock(&inode->i_lock);
  896. mine = cap->issued | cap->implemented;
  897. used = __ceph_caps_used(ci);
  898. oissued = __ceph_caps_issued_other(ci, cap);
  899. dout("trim_caps_cb %p cap %p mine %s oissued %s used %s\n",
  900. inode, cap, ceph_cap_string(mine), ceph_cap_string(oissued),
  901. ceph_cap_string(used));
  902. if (ci->i_dirty_caps)
  903. goto out; /* dirty caps */
  904. if ((used & ~oissued) & mine)
  905. goto out; /* we need these caps */
  906. session->s_trim_caps--;
  907. if (oissued) {
  908. /* we aren't the only cap.. just remove us */
  909. __ceph_remove_cap(cap);
  910. } else {
  911. /* try to drop referring dentries */
  912. spin_unlock(&inode->i_lock);
  913. d_prune_aliases(inode);
  914. dout("trim_caps_cb %p cap %p pruned, count now %d\n",
  915. inode, cap, atomic_read(&inode->i_count));
  916. return 0;
  917. }
  918. out:
  919. spin_unlock(&inode->i_lock);
  920. return 0;
  921. }
  922. /*
  923. * Trim session cap count down to some max number.
  924. */
  925. static int trim_caps(struct ceph_mds_client *mdsc,
  926. struct ceph_mds_session *session,
  927. int max_caps)
  928. {
  929. int trim_caps = session->s_nr_caps - max_caps;
  930. dout("trim_caps mds%d start: %d / %d, trim %d\n",
  931. session->s_mds, session->s_nr_caps, max_caps, trim_caps);
  932. if (trim_caps > 0) {
  933. session->s_trim_caps = trim_caps;
  934. iterate_session_caps(session, trim_caps_cb, session);
  935. dout("trim_caps mds%d done: %d / %d, trimmed %d\n",
  936. session->s_mds, session->s_nr_caps, max_caps,
  937. trim_caps - session->s_trim_caps);
  938. session->s_trim_caps = 0;
  939. }
  940. return 0;
  941. }
  942. /*
  943. * Allocate cap_release messages. If there is a partially full message
  944. * in the queue, try to allocate enough to cover it's remainder, so that
  945. * we can send it immediately.
  946. *
  947. * Called under s_mutex.
  948. */
  949. static int add_cap_releases(struct ceph_mds_client *mdsc,
  950. struct ceph_mds_session *session,
  951. int extra)
  952. {
  953. struct ceph_msg *msg;
  954. struct ceph_mds_cap_release *head;
  955. int err = -ENOMEM;
  956. if (extra < 0)
  957. extra = mdsc->client->mount_args->cap_release_safety;
  958. spin_lock(&session->s_cap_lock);
  959. if (!list_empty(&session->s_cap_releases)) {
  960. msg = list_first_entry(&session->s_cap_releases,
  961. struct ceph_msg,
  962. list_head);
  963. head = msg->front.iov_base;
  964. extra += CEPH_CAPS_PER_RELEASE - le32_to_cpu(head->num);
  965. }
  966. while (session->s_num_cap_releases < session->s_nr_caps + extra) {
  967. spin_unlock(&session->s_cap_lock);
  968. msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPRELEASE, PAGE_CACHE_SIZE,
  969. GFP_NOFS);
  970. if (!msg)
  971. goto out_unlocked;
  972. dout("add_cap_releases %p msg %p now %d\n", session, msg,
  973. (int)msg->front.iov_len);
  974. head = msg->front.iov_base;
  975. head->num = cpu_to_le32(0);
  976. msg->front.iov_len = sizeof(*head);
  977. spin_lock(&session->s_cap_lock);
  978. list_add(&msg->list_head, &session->s_cap_releases);
  979. session->s_num_cap_releases += CEPH_CAPS_PER_RELEASE;
  980. }
  981. if (!list_empty(&session->s_cap_releases)) {
  982. msg = list_first_entry(&session->s_cap_releases,
  983. struct ceph_msg,
  984. list_head);
  985. head = msg->front.iov_base;
  986. if (head->num) {
  987. dout(" queueing non-full %p (%d)\n", msg,
  988. le32_to_cpu(head->num));
  989. list_move_tail(&msg->list_head,
  990. &session->s_cap_releases_done);
  991. session->s_num_cap_releases -=
  992. CEPH_CAPS_PER_RELEASE - le32_to_cpu(head->num);
  993. }
  994. }
  995. err = 0;
  996. spin_unlock(&session->s_cap_lock);
  997. out_unlocked:
  998. return err;
  999. }
  1000. /*
  1001. * flush all dirty inode data to disk.
  1002. *
  1003. * returns true if we've flushed through want_flush_seq
  1004. */
  1005. static int check_cap_flush(struct ceph_mds_client *mdsc, u64 want_flush_seq)
  1006. {
  1007. int mds, ret = 1;
  1008. dout("check_cap_flush want %lld\n", want_flush_seq);
  1009. mutex_lock(&mdsc->mutex);
  1010. for (mds = 0; ret && mds < mdsc->max_sessions; mds++) {
  1011. struct ceph_mds_session *session = mdsc->sessions[mds];
  1012. if (!session)
  1013. continue;
  1014. get_session(session);
  1015. mutex_unlock(&mdsc->mutex);
  1016. mutex_lock(&session->s_mutex);
  1017. if (!list_empty(&session->s_cap_flushing)) {
  1018. struct ceph_inode_info *ci =
  1019. list_entry(session->s_cap_flushing.next,
  1020. struct ceph_inode_info,
  1021. i_flushing_item);
  1022. struct inode *inode = &ci->vfs_inode;
  1023. spin_lock(&inode->i_lock);
  1024. if (ci->i_cap_flush_seq <= want_flush_seq) {
  1025. dout("check_cap_flush still flushing %p "
  1026. "seq %lld <= %lld to mds%d\n", inode,
  1027. ci->i_cap_flush_seq, want_flush_seq,
  1028. session->s_mds);
  1029. ret = 0;
  1030. }
  1031. spin_unlock(&inode->i_lock);
  1032. }
  1033. mutex_unlock(&session->s_mutex);
  1034. ceph_put_mds_session(session);
  1035. if (!ret)
  1036. return ret;
  1037. mutex_lock(&mdsc->mutex);
  1038. }
  1039. mutex_unlock(&mdsc->mutex);
  1040. dout("check_cap_flush ok, flushed thru %lld\n", want_flush_seq);
  1041. return ret;
  1042. }
  1043. /*
  1044. * called under s_mutex
  1045. */
  1046. static void send_cap_releases(struct ceph_mds_client *mdsc,
  1047. struct ceph_mds_session *session)
  1048. {
  1049. struct ceph_msg *msg;
  1050. dout("send_cap_releases mds%d\n", session->s_mds);
  1051. spin_lock(&session->s_cap_lock);
  1052. while (!list_empty(&session->s_cap_releases_done)) {
  1053. msg = list_first_entry(&session->s_cap_releases_done,
  1054. struct ceph_msg, list_head);
  1055. list_del_init(&msg->list_head);
  1056. spin_unlock(&session->s_cap_lock);
  1057. msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
  1058. dout("send_cap_releases mds%d %p\n", session->s_mds, msg);
  1059. ceph_con_send(&session->s_con, msg);
  1060. spin_lock(&session->s_cap_lock);
  1061. }
  1062. spin_unlock(&session->s_cap_lock);
  1063. }
  1064. static void discard_cap_releases(struct ceph_mds_client *mdsc,
  1065. struct ceph_mds_session *session)
  1066. {
  1067. struct ceph_msg *msg;
  1068. struct ceph_mds_cap_release *head;
  1069. unsigned num;
  1070. dout("discard_cap_releases mds%d\n", session->s_mds);
  1071. spin_lock(&session->s_cap_lock);
  1072. /* zero out the in-progress message */
  1073. msg = list_first_entry(&session->s_cap_releases,
  1074. struct ceph_msg, list_head);
  1075. head = msg->front.iov_base;
  1076. num = le32_to_cpu(head->num);
  1077. dout("discard_cap_releases mds%d %p %u\n", session->s_mds, msg, num);
  1078. head->num = cpu_to_le32(0);
  1079. session->s_num_cap_releases += num;
  1080. /* requeue completed messages */
  1081. while (!list_empty(&session->s_cap_releases_done)) {
  1082. msg = list_first_entry(&session->s_cap_releases_done,
  1083. struct ceph_msg, list_head);
  1084. list_del_init(&msg->list_head);
  1085. head = msg->front.iov_base;
  1086. num = le32_to_cpu(head->num);
  1087. dout("discard_cap_releases mds%d %p %u\n", session->s_mds, msg,
  1088. num);
  1089. session->s_num_cap_releases += num;
  1090. head->num = cpu_to_le32(0);
  1091. msg->front.iov_len = sizeof(*head);
  1092. list_add(&msg->list_head, &session->s_cap_releases);
  1093. }
  1094. spin_unlock(&session->s_cap_lock);
  1095. }
  1096. /*
  1097. * requests
  1098. */
  1099. /*
  1100. * Create an mds request.
  1101. */
  1102. struct ceph_mds_request *
  1103. ceph_mdsc_create_request(struct ceph_mds_client *mdsc, int op, int mode)
  1104. {
  1105. struct ceph_mds_request *req = kzalloc(sizeof(*req), GFP_NOFS);
  1106. if (!req)
  1107. return ERR_PTR(-ENOMEM);
  1108. mutex_init(&req->r_fill_mutex);
  1109. req->r_started = jiffies;
  1110. req->r_resend_mds = -1;
  1111. INIT_LIST_HEAD(&req->r_unsafe_dir_item);
  1112. req->r_fmode = -1;
  1113. kref_init(&req->r_kref);
  1114. INIT_LIST_HEAD(&req->r_wait);
  1115. init_completion(&req->r_completion);
  1116. init_completion(&req->r_safe_completion);
  1117. INIT_LIST_HEAD(&req->r_unsafe_item);
  1118. req->r_op = op;
  1119. req->r_direct_mode = mode;
  1120. return req;
  1121. }
  1122. /*
  1123. * return oldest (lowest) request, tid in request tree, 0 if none.
  1124. *
  1125. * called under mdsc->mutex.
  1126. */
  1127. static struct ceph_mds_request *__get_oldest_req(struct ceph_mds_client *mdsc)
  1128. {
  1129. if (RB_EMPTY_ROOT(&mdsc->request_tree))
  1130. return NULL;
  1131. return rb_entry(rb_first(&mdsc->request_tree),
  1132. struct ceph_mds_request, r_node);
  1133. }
  1134. static u64 __get_oldest_tid(struct ceph_mds_client *mdsc)
  1135. {
  1136. struct ceph_mds_request *req = __get_oldest_req(mdsc);
  1137. if (req)
  1138. return req->r_tid;
  1139. return 0;
  1140. }
  1141. /*
  1142. * Build a dentry's path. Allocate on heap; caller must kfree. Based
  1143. * on build_path_from_dentry in fs/cifs/dir.c.
  1144. *
  1145. * If @stop_on_nosnap, generate path relative to the first non-snapped
  1146. * inode.
  1147. *
  1148. * Encode hidden .snap dirs as a double /, i.e.
  1149. * foo/.snap/bar -> foo//bar
  1150. */
  1151. char *ceph_mdsc_build_path(struct dentry *dentry, int *plen, u64 *base,
  1152. int stop_on_nosnap)
  1153. {
  1154. struct dentry *temp;
  1155. char *path;
  1156. int len, pos;
  1157. if (dentry == NULL)
  1158. return ERR_PTR(-EINVAL);
  1159. retry:
  1160. len = 0;
  1161. for (temp = dentry; !IS_ROOT(temp);) {
  1162. struct inode *inode = temp->d_inode;
  1163. if (inode && ceph_snap(inode) == CEPH_SNAPDIR)
  1164. len++; /* slash only */
  1165. else if (stop_on_nosnap && inode &&
  1166. ceph_snap(inode) == CEPH_NOSNAP)
  1167. break;
  1168. else
  1169. len += 1 + temp->d_name.len;
  1170. temp = temp->d_parent;
  1171. if (temp == NULL) {
  1172. pr_err("build_path corrupt dentry %p\n", dentry);
  1173. return ERR_PTR(-EINVAL);
  1174. }
  1175. }
  1176. if (len)
  1177. len--; /* no leading '/' */
  1178. path = kmalloc(len+1, GFP_NOFS);
  1179. if (path == NULL)
  1180. return ERR_PTR(-ENOMEM);
  1181. pos = len;
  1182. path[pos] = 0; /* trailing null */
  1183. for (temp = dentry; !IS_ROOT(temp) && pos != 0; ) {
  1184. struct inode *inode = temp->d_inode;
  1185. if (inode && ceph_snap(inode) == CEPH_SNAPDIR) {
  1186. dout("build_path path+%d: %p SNAPDIR\n",
  1187. pos, temp);
  1188. } else if (stop_on_nosnap && inode &&
  1189. ceph_snap(inode) == CEPH_NOSNAP) {
  1190. break;
  1191. } else {
  1192. pos -= temp->d_name.len;
  1193. if (pos < 0)
  1194. break;
  1195. strncpy(path + pos, temp->d_name.name,
  1196. temp->d_name.len);
  1197. }
  1198. if (pos)
  1199. path[--pos] = '/';
  1200. temp = temp->d_parent;
  1201. if (temp == NULL) {
  1202. pr_err("build_path corrupt dentry\n");
  1203. kfree(path);
  1204. return ERR_PTR(-EINVAL);
  1205. }
  1206. }
  1207. if (pos != 0) {
  1208. pr_err("build_path did not end path lookup where "
  1209. "expected, namelen is %d, pos is %d\n", len, pos);
  1210. /* presumably this is only possible if racing with a
  1211. rename of one of the parent directories (we can not
  1212. lock the dentries above us to prevent this, but
  1213. retrying should be harmless) */
  1214. kfree(path);
  1215. goto retry;
  1216. }
  1217. *base = ceph_ino(temp->d_inode);
  1218. *plen = len;
  1219. dout("build_path on %p %d built %llx '%.*s'\n",
  1220. dentry, atomic_read(&dentry->d_count), *base, len, path);
  1221. return path;
  1222. }
  1223. static int build_dentry_path(struct dentry *dentry,
  1224. const char **ppath, int *ppathlen, u64 *pino,
  1225. int *pfreepath)
  1226. {
  1227. char *path;
  1228. if (ceph_snap(dentry->d_parent->d_inode) == CEPH_NOSNAP) {
  1229. *pino = ceph_ino(dentry->d_parent->d_inode);
  1230. *ppath = dentry->d_name.name;
  1231. *ppathlen = dentry->d_name.len;
  1232. return 0;
  1233. }
  1234. path = ceph_mdsc_build_path(dentry, ppathlen, pino, 1);
  1235. if (IS_ERR(path))
  1236. return PTR_ERR(path);
  1237. *ppath = path;
  1238. *pfreepath = 1;
  1239. return 0;
  1240. }
  1241. static int build_inode_path(struct inode *inode,
  1242. const char **ppath, int *ppathlen, u64 *pino,
  1243. int *pfreepath)
  1244. {
  1245. struct dentry *dentry;
  1246. char *path;
  1247. if (ceph_snap(inode) == CEPH_NOSNAP) {
  1248. *pino = ceph_ino(inode);
  1249. *ppathlen = 0;
  1250. return 0;
  1251. }
  1252. dentry = d_find_alias(inode);
  1253. path = ceph_mdsc_build_path(dentry, ppathlen, pino, 1);
  1254. dput(dentry);
  1255. if (IS_ERR(path))
  1256. return PTR_ERR(path);
  1257. *ppath = path;
  1258. *pfreepath = 1;
  1259. return 0;
  1260. }
  1261. /*
  1262. * request arguments may be specified via an inode *, a dentry *, or
  1263. * an explicit ino+path.
  1264. */
  1265. static int set_request_path_attr(struct inode *rinode, struct dentry *rdentry,
  1266. const char *rpath, u64 rino,
  1267. const char **ppath, int *pathlen,
  1268. u64 *ino, int *freepath)
  1269. {
  1270. int r = 0;
  1271. if (rinode) {
  1272. r = build_inode_path(rinode, ppath, pathlen, ino, freepath);
  1273. dout(" inode %p %llx.%llx\n", rinode, ceph_ino(rinode),
  1274. ceph_snap(rinode));
  1275. } else if (rdentry) {
  1276. r = build_dentry_path(rdentry, ppath, pathlen, ino, freepath);
  1277. dout(" dentry %p %llx/%.*s\n", rdentry, *ino, *pathlen,
  1278. *ppath);
  1279. } else if (rpath) {
  1280. *ino = rino;
  1281. *ppath = rpath;
  1282. *pathlen = strlen(rpath);
  1283. dout(" path %.*s\n", *pathlen, rpath);
  1284. }
  1285. return r;
  1286. }
  1287. /*
  1288. * called under mdsc->mutex
  1289. */
  1290. static struct ceph_msg *create_request_message(struct ceph_mds_client *mdsc,
  1291. struct ceph_mds_request *req,
  1292. int mds)
  1293. {
  1294. struct ceph_msg *msg;
  1295. struct ceph_mds_request_head *head;
  1296. const char *path1 = NULL;
  1297. const char *path2 = NULL;
  1298. u64 ino1 = 0, ino2 = 0;
  1299. int pathlen1 = 0, pathlen2 = 0;
  1300. int freepath1 = 0, freepath2 = 0;
  1301. int len;
  1302. u16 releases;
  1303. void *p, *end;
  1304. int ret;
  1305. ret = set_request_path_attr(req->r_inode, req->r_dentry,
  1306. req->r_path1, req->r_ino1.ino,
  1307. &path1, &pathlen1, &ino1, &freepath1);
  1308. if (ret < 0) {
  1309. msg = ERR_PTR(ret);
  1310. goto out;
  1311. }
  1312. ret = set_request_path_attr(NULL, req->r_old_dentry,
  1313. req->r_path2, req->r_ino2.ino,
  1314. &path2, &pathlen2, &ino2, &freepath2);
  1315. if (ret < 0) {
  1316. msg = ERR_PTR(ret);
  1317. goto out_free1;
  1318. }
  1319. len = sizeof(*head) +
  1320. pathlen1 + pathlen2 + 2*(1 + sizeof(u32) + sizeof(u64));
  1321. /* calculate (max) length for cap releases */
  1322. len += sizeof(struct ceph_mds_request_release) *
  1323. (!!req->r_inode_drop + !!req->r_dentry_drop +
  1324. !!req->r_old_inode_drop + !!req->r_old_dentry_drop);
  1325. if (req->r_dentry_drop)
  1326. len += req->r_dentry->d_name.len;
  1327. if (req->r_old_dentry_drop)
  1328. len += req->r_old_dentry->d_name.len;
  1329. msg = ceph_msg_new(CEPH_MSG_CLIENT_REQUEST, len, GFP_NOFS);
  1330. if (!msg) {
  1331. msg = ERR_PTR(-ENOMEM);
  1332. goto out_free2;
  1333. }
  1334. msg->hdr.tid = cpu_to_le64(req->r_tid);
  1335. head = msg->front.iov_base;
  1336. p = msg->front.iov_base + sizeof(*head);
  1337. end = msg->front.iov_base + msg->front.iov_len;
  1338. head->mdsmap_epoch = cpu_to_le32(mdsc->mdsmap->m_epoch);
  1339. head->op = cpu_to_le32(req->r_op);
  1340. head->caller_uid = cpu_to_le32(current_fsuid());
  1341. head->caller_gid = cpu_to_le32(current_fsgid());
  1342. head->args = req->r_args;
  1343. ceph_encode_filepath(&p, end, ino1, path1);
  1344. ceph_encode_filepath(&p, end, ino2, path2);
  1345. /* cap releases */
  1346. releases = 0;
  1347. if (req->r_inode_drop)
  1348. releases += ceph_encode_inode_release(&p,
  1349. req->r_inode ? req->r_inode : req->r_dentry->d_inode,
  1350. mds, req->r_inode_drop, req->r_inode_unless, 0);
  1351. if (req->r_dentry_drop)
  1352. releases += ceph_encode_dentry_release(&p, req->r_dentry,
  1353. mds, req->r_dentry_drop, req->r_dentry_unless);
  1354. if (req->r_old_dentry_drop)
  1355. releases += ceph_encode_dentry_release(&p, req->r_old_dentry,
  1356. mds, req->r_old_dentry_drop, req->r_old_dentry_unless);
  1357. if (req->r_old_inode_drop)
  1358. releases += ceph_encode_inode_release(&p,
  1359. req->r_old_dentry->d_inode,
  1360. mds, req->r_old_inode_drop, req->r_old_inode_unless, 0);
  1361. head->num_releases = cpu_to_le16(releases);
  1362. BUG_ON(p > end);
  1363. msg->front.iov_len = p - msg->front.iov_base;
  1364. msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
  1365. msg->pages = req->r_pages;
  1366. msg->nr_pages = req->r_num_pages;
  1367. msg->hdr.data_len = cpu_to_le32(req->r_data_len);
  1368. msg->hdr.data_off = cpu_to_le16(0);
  1369. out_free2:
  1370. if (freepath2)
  1371. kfree((char *)path2);
  1372. out_free1:
  1373. if (freepath1)
  1374. kfree((char *)path1);
  1375. out:
  1376. return msg;
  1377. }
  1378. /*
  1379. * called under mdsc->mutex if error, under no mutex if
  1380. * success.
  1381. */
  1382. static void complete_request(struct ceph_mds_client *mdsc,
  1383. struct ceph_mds_request *req)
  1384. {
  1385. if (req->r_callback)
  1386. req->r_callback(mdsc, req);
  1387. else
  1388. complete(&req->r_completion);
  1389. }
  1390. /*
  1391. * called under mdsc->mutex
  1392. */
  1393. static int __prepare_send_request(struct ceph_mds_client *mdsc,
  1394. struct ceph_mds_request *req,
  1395. int mds)
  1396. {
  1397. struct ceph_mds_request_head *rhead;
  1398. struct ceph_msg *msg;
  1399. int flags = 0;
  1400. req->r_mds = mds;
  1401. req->r_attempts++;
  1402. dout("prepare_send_request %p tid %lld %s (attempt %d)\n", req,
  1403. req->r_tid, ceph_mds_op_name(req->r_op), req->r_attempts);
  1404. if (req->r_request) {
  1405. ceph_msg_put(req->r_request);
  1406. req->r_request = NULL;
  1407. }
  1408. msg = create_request_message(mdsc, req, mds);
  1409. if (IS_ERR(msg)) {
  1410. req->r_err = PTR_ERR(msg);
  1411. complete_request(mdsc, req);
  1412. return PTR_ERR(msg);
  1413. }
  1414. req->r_request = msg;
  1415. rhead = msg->front.iov_base;
  1416. rhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
  1417. if (req->r_got_unsafe)
  1418. flags |= CEPH_MDS_FLAG_REPLAY;
  1419. if (req->r_locked_dir)
  1420. flags |= CEPH_MDS_FLAG_WANT_DENTRY;
  1421. rhead->flags = cpu_to_le32(flags);
  1422. rhead->num_fwd = req->r_num_fwd;
  1423. rhead->num_retry = req->r_attempts - 1;
  1424. dout(" r_locked_dir = %p\n", req->r_locked_dir);
  1425. if (req->r_target_inode && req->r_got_unsafe)
  1426. rhead->ino = cpu_to_le64(ceph_ino(req->r_target_inode));
  1427. else
  1428. rhead->ino = 0;
  1429. return 0;
  1430. }
  1431. /*
  1432. * send request, or put it on the appropriate wait list.
  1433. */
  1434. static int __do_request(struct ceph_mds_client *mdsc,
  1435. struct ceph_mds_request *req)
  1436. {
  1437. struct ceph_mds_session *session = NULL;
  1438. int mds = -1;
  1439. int err = -EAGAIN;
  1440. if (req->r_err || req->r_got_result)
  1441. goto out;
  1442. if (req->r_timeout &&
  1443. time_after_eq(jiffies, req->r_started + req->r_timeout)) {
  1444. dout("do_request timed out\n");
  1445. err = -EIO;
  1446. goto finish;
  1447. }
  1448. mds = __choose_mds(mdsc, req);
  1449. if (mds < 0 ||
  1450. ceph_mdsmap_get_state(mdsc->mdsmap, mds) < CEPH_MDS_STATE_ACTIVE) {
  1451. dout("do_request no mds or not active, waiting for map\n");
  1452. list_add(&req->r_wait, &mdsc->waiting_for_map);
  1453. goto out;
  1454. }
  1455. /* get, open session */
  1456. session = __ceph_lookup_mds_session(mdsc, mds);
  1457. if (!session) {
  1458. session = register_session(mdsc, mds);
  1459. if (IS_ERR(session)) {
  1460. err = PTR_ERR(session);
  1461. goto finish;
  1462. }
  1463. }
  1464. dout("do_request mds%d session %p state %s\n", mds, session,
  1465. session_state_name(session->s_state));
  1466. if (session->s_state != CEPH_MDS_SESSION_OPEN &&
  1467. session->s_state != CEPH_MDS_SESSION_HUNG) {
  1468. if (session->s_state == CEPH_MDS_SESSION_NEW ||
  1469. session->s_state == CEPH_MDS_SESSION_CLOSING)
  1470. __open_session(mdsc, session);
  1471. list_add(&req->r_wait, &session->s_waiting);
  1472. goto out_session;
  1473. }
  1474. /* send request */
  1475. req->r_session = get_session(session);
  1476. req->r_resend_mds = -1; /* forget any previous mds hint */
  1477. if (req->r_request_started == 0) /* note request start time */
  1478. req->r_request_started = jiffies;
  1479. err = __prepare_send_request(mdsc, req, mds);
  1480. if (!err) {
  1481. ceph_msg_get(req->r_request);
  1482. ceph_con_send(&session->s_con, req->r_request);
  1483. }
  1484. out_session:
  1485. ceph_put_mds_session(session);
  1486. out:
  1487. return err;
  1488. finish:
  1489. req->r_err = err;
  1490. complete_request(mdsc, req);
  1491. goto out;
  1492. }
  1493. /*
  1494. * called under mdsc->mutex
  1495. */
  1496. static void __wake_requests(struct ceph_mds_client *mdsc,
  1497. struct list_head *head)
  1498. {
  1499. struct ceph_mds_request *req, *nreq;
  1500. list_for_each_entry_safe(req, nreq, head, r_wait) {
  1501. list_del_init(&req->r_wait);
  1502. __do_request(mdsc, req);
  1503. }
  1504. }
  1505. /*
  1506. * Wake up threads with requests pending for @mds, so that they can
  1507. * resubmit their requests to a possibly different mds.
  1508. */
  1509. static void kick_requests(struct ceph_mds_client *mdsc, int mds)
  1510. {
  1511. struct ceph_mds_request *req;
  1512. struct rb_node *p;
  1513. dout("kick_requests mds%d\n", mds);
  1514. for (p = rb_first(&mdsc->request_tree); p; p = rb_next(p)) {
  1515. req = rb_entry(p, struct ceph_mds_request, r_node);
  1516. if (req->r_got_unsafe)
  1517. continue;
  1518. if (req->r_session &&
  1519. req->r_session->s_mds == mds) {
  1520. dout(" kicking tid %llu\n", req->r_tid);
  1521. put_request_session(req);
  1522. __do_request(mdsc, req);
  1523. }
  1524. }
  1525. }
  1526. void ceph_mdsc_submit_request(struct ceph_mds_client *mdsc,
  1527. struct ceph_mds_request *req)
  1528. {
  1529. dout("submit_request on %p\n", req);
  1530. mutex_lock(&mdsc->mutex);
  1531. __register_request(mdsc, req, NULL);
  1532. __do_request(mdsc, req);
  1533. mutex_unlock(&mdsc->mutex);
  1534. }
  1535. /*
  1536. * Synchrously perform an mds request. Take care of all of the
  1537. * session setup, forwarding, retry details.
  1538. */
  1539. int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
  1540. struct inode *dir,
  1541. struct ceph_mds_request *req)
  1542. {
  1543. int err;
  1544. dout("do_request on %p\n", req);
  1545. /* take CAP_PIN refs for r_inode, r_locked_dir, r_old_dentry */
  1546. if (req->r_inode)
  1547. ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
  1548. if (req->r_locked_dir)
  1549. ceph_get_cap_refs(ceph_inode(req->r_locked_dir), CEPH_CAP_PIN);
  1550. if (req->r_old_dentry)
  1551. ceph_get_cap_refs(
  1552. ceph_inode(req->r_old_dentry->d_parent->d_inode),
  1553. CEPH_CAP_PIN);
  1554. /* issue */
  1555. mutex_lock(&mdsc->mutex);
  1556. __register_request(mdsc, req, dir);
  1557. __do_request(mdsc, req);
  1558. if (req->r_err) {
  1559. err = req->r_err;
  1560. __unregister_request(mdsc, req);
  1561. dout("do_request early error %d\n", err);
  1562. goto out;
  1563. }
  1564. /* wait */
  1565. mutex_unlock(&mdsc->mutex);
  1566. dout("do_request waiting\n");
  1567. if (req->r_timeout) {
  1568. err = (long)wait_for_completion_interruptible_timeout(
  1569. &req->r_completion, req->r_timeout);
  1570. if (err == 0)
  1571. err = -EIO;
  1572. } else {
  1573. err = wait_for_completion_interruptible(&req->r_completion);
  1574. }
  1575. dout("do_request waited, got %d\n", err);
  1576. mutex_lock(&mdsc->mutex);
  1577. /* only abort if we didn't race with a real reply */
  1578. if (req->r_got_result) {
  1579. err = le32_to_cpu(req->r_reply_info.head->result);
  1580. } else if (err < 0) {
  1581. dout("aborted request %lld with %d\n", req->r_tid, err);
  1582. /*
  1583. * ensure we aren't running concurrently with
  1584. * ceph_fill_trace or ceph_readdir_prepopulate, which
  1585. * rely on locks (dir mutex) held by our caller.
  1586. */
  1587. mutex_lock(&req->r_fill_mutex);
  1588. req->r_err = err;
  1589. req->r_aborted = true;
  1590. mutex_unlock(&req->r_fill_mutex);
  1591. if (req->r_locked_dir &&
  1592. (req->r_op & CEPH_MDS_OP_WRITE))
  1593. ceph_invalidate_dir_request(req);
  1594. } else {
  1595. err = req->r_err;
  1596. }
  1597. out:
  1598. mutex_unlock(&mdsc->mutex);
  1599. dout("do_request %p done, result %d\n", req, err);
  1600. return err;
  1601. }
  1602. /*
  1603. * Invalidate dir I_COMPLETE, dentry lease state on an aborted MDS
  1604. * namespace request.
  1605. */
  1606. void ceph_invalidate_dir_request(struct ceph_mds_request *req)
  1607. {
  1608. struct inode *inode = req->r_locked_dir;
  1609. struct ceph_inode_info *ci = ceph_inode(inode);
  1610. dout("invalidate_dir_request %p (I_COMPLETE, lease(s))\n", inode);
  1611. spin_lock(&inode->i_lock);
  1612. ci->i_ceph_flags &= ~CEPH_I_COMPLETE;
  1613. ci->i_release_count++;
  1614. spin_unlock(&inode->i_lock);
  1615. if (req->r_dentry)
  1616. ceph_invalidate_dentry_lease(req->r_dentry);
  1617. if (req->r_old_dentry)
  1618. ceph_invalidate_dentry_lease(req->r_old_dentry);
  1619. }
  1620. /*
  1621. * Handle mds reply.
  1622. *
  1623. * We take the session mutex and parse and process the reply immediately.
  1624. * This preserves the logical ordering of replies, capabilities, etc., sent
  1625. * by the MDS as they are applied to our local cache.
  1626. */
  1627. static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
  1628. {
  1629. struct ceph_mds_client *mdsc = session->s_mdsc;
  1630. struct ceph_mds_request *req;
  1631. struct ceph_mds_reply_head *head = msg->front.iov_base;
  1632. struct ceph_mds_reply_info_parsed *rinfo; /* parsed reply info */
  1633. u64 tid;
  1634. int err, result;
  1635. int mds = session->s_mds;
  1636. if (msg->front.iov_len < sizeof(*head)) {
  1637. pr_err("mdsc_handle_reply got corrupt (short) reply\n");
  1638. ceph_msg_dump(msg);
  1639. return;
  1640. }
  1641. /* get request, session */
  1642. tid = le64_to_cpu(msg->hdr.tid);
  1643. mutex_lock(&mdsc->mutex);
  1644. req = __lookup_request(mdsc, tid);
  1645. if (!req) {
  1646. dout("handle_reply on unknown tid %llu\n", tid);
  1647. mutex_unlock(&mdsc->mutex);
  1648. return;
  1649. }
  1650. dout("handle_reply %p\n", req);
  1651. /* correct session? */
  1652. if (req->r_session != session) {
  1653. pr_err("mdsc_handle_reply got %llu on session mds%d"
  1654. " not mds%d\n", tid, session->s_mds,
  1655. req->r_session ? req->r_session->s_mds : -1);
  1656. mutex_unlock(&mdsc->mutex);
  1657. goto out;
  1658. }
  1659. /* dup? */
  1660. if ((req->r_got_unsafe && !head->safe) ||
  1661. (req->r_got_safe && head->safe)) {
  1662. pr_warning("got a dup %s reply on %llu from mds%d\n",
  1663. head->safe ? "safe" : "unsafe", tid, mds);
  1664. mutex_unlock(&mdsc->mutex);
  1665. goto out;
  1666. }
  1667. if (req->r_got_safe && !head->safe) {
  1668. pr_warning("got unsafe after safe on %llu from mds%d\n",
  1669. tid, mds);
  1670. mutex_unlock(&mdsc->mutex);
  1671. goto out;
  1672. }
  1673. result = le32_to_cpu(head->result);
  1674. /*
  1675. * Tolerate 2 consecutive ESTALEs from the same mds.
  1676. * FIXME: we should be looking at the cap migrate_seq.
  1677. */
  1678. if (result == -ESTALE) {
  1679. req->r_direct_mode = USE_AUTH_MDS;
  1680. req->r_num_stale++;
  1681. if (req->r_num_stale <= 2) {
  1682. __do_request(mdsc, req);
  1683. mutex_unlock(&mdsc->mutex);
  1684. goto out;
  1685. }
  1686. } else {
  1687. req->r_num_stale = 0;
  1688. }
  1689. if (head->safe) {
  1690. req->r_got_safe = true;
  1691. __unregister_request(mdsc, req);
  1692. complete(&req->r_safe_completion);
  1693. if (req->r_got_unsafe) {
  1694. /*
  1695. * We already handled the unsafe response, now do the
  1696. * cleanup. No need to examine the response; the MDS
  1697. * doesn't include any result info in the safe
  1698. * response. And even if it did, there is nothing
  1699. * useful we could do with a revised return value.
  1700. */
  1701. dout("got safe reply %llu, mds%d\n", tid, mds);
  1702. list_del_init(&req->r_unsafe_item);
  1703. /* last unsafe request during umount? */
  1704. if (mdsc->stopping && !__get_oldest_req(mdsc))
  1705. complete(&mdsc->safe_umount_waiters);
  1706. mutex_unlock(&mdsc->mutex);
  1707. goto out;
  1708. }
  1709. } else {
  1710. req->r_got_unsafe = true;
  1711. list_add_tail(&req->r_unsafe_item, &req->r_session->s_unsafe);
  1712. }
  1713. dout("handle_reply tid %lld result %d\n", tid, result);
  1714. rinfo = &req->r_reply_info;
  1715. err = parse_reply_info(msg, rinfo);
  1716. mutex_unlock(&mdsc->mutex);
  1717. mutex_lock(&session->s_mutex);
  1718. if (err < 0) {
  1719. pr_err("mdsc_handle_reply got corrupt reply mds%d\n", mds);
  1720. ceph_msg_dump(msg);
  1721. goto out_err;
  1722. }
  1723. /* snap trace */
  1724. if (rinfo->snapblob_len) {
  1725. down_write(&mdsc->snap_rwsem);
  1726. ceph_update_snap_trace(mdsc, rinfo->snapblob,
  1727. rinfo->snapblob + rinfo->snapblob_len,
  1728. le32_to_cpu(head->op) == CEPH_MDS_OP_RMSNAP);
  1729. downgrade_write(&mdsc->snap_rwsem);
  1730. } else {
  1731. down_read(&mdsc->snap_rwsem);
  1732. }
  1733. /* insert trace into our cache */
  1734. mutex_lock(&req->r_fill_mutex);
  1735. err = ceph_fill_trace(mdsc->client->sb, req, req->r_session);
  1736. if (err == 0) {
  1737. if (result == 0 && rinfo->dir_nr)
  1738. ceph_readdir_prepopulate(req, req->r_session);
  1739. ceph_unreserve_caps(&req->r_caps_reservation);
  1740. }
  1741. mutex_unlock(&req->r_fill_mutex);
  1742. up_read(&mdsc->snap_rwsem);
  1743. out_err:
  1744. mutex_lock(&mdsc->mutex);
  1745. if (!req->r_aborted) {
  1746. if (err) {
  1747. req->r_err = err;
  1748. } else {
  1749. req->r_reply = msg;
  1750. ceph_msg_get(msg);
  1751. req->r_got_result = true;
  1752. }
  1753. } else {
  1754. dout("reply arrived after request %lld was aborted\n", tid);
  1755. }
  1756. mutex_unlock(&mdsc->mutex);
  1757. add_cap_releases(mdsc, req->r_session, -1);
  1758. mutex_unlock(&session->s_mutex);
  1759. /* kick calling process */
  1760. complete_request(mdsc, req);
  1761. out:
  1762. ceph_mdsc_put_request(req);
  1763. return;
  1764. }
  1765. /*
  1766. * handle mds notification that our request has been forwarded.
  1767. */
  1768. static void handle_forward(struct ceph_mds_client *mdsc,
  1769. struct ceph_mds_session *session,
  1770. struct ceph_msg *msg)
  1771. {
  1772. struct ceph_mds_request *req;
  1773. u64 tid = le64_to_cpu(msg->hdr.tid);
  1774. u32 next_mds;
  1775. u32 fwd_seq;
  1776. int err = -EINVAL;
  1777. void *p = msg->front.iov_base;
  1778. void *end = p + msg->front.iov_len;
  1779. ceph_decode_need(&p, end, 2*sizeof(u32), bad);
  1780. next_mds = ceph_decode_32(&p);
  1781. fwd_seq = ceph_decode_32(&p);
  1782. mutex_lock(&mdsc->mutex);
  1783. req = __lookup_request(mdsc, tid);
  1784. if (!req) {
  1785. dout("forward %llu to mds%d - req dne\n", tid, next_mds);
  1786. goto out; /* dup reply? */
  1787. }
  1788. if (fwd_seq <= req->r_num_fwd) {
  1789. dout("forward %llu to mds%d - old seq %d <= %d\n",
  1790. tid, next_mds, req->r_num_fwd, fwd_seq);
  1791. } else {
  1792. /* resend. forward race not possible; mds would drop */
  1793. dout("forward %llu to mds%d (we resend)\n", tid, next_mds);
  1794. req->r_num_fwd = fwd_seq;
  1795. req->r_resend_mds = next_mds;
  1796. put_request_session(req);
  1797. __do_request(mdsc, req);
  1798. }
  1799. ceph_mdsc_put_request(req);
  1800. out:
  1801. mutex_unlock(&mdsc->mutex);
  1802. return;
  1803. bad:
  1804. pr_err("mdsc_handle_forward decode error err=%d\n", err);
  1805. }
  1806. /*
  1807. * handle a mds session control message
  1808. */
  1809. static void handle_session(struct ceph_mds_session *session,
  1810. struct ceph_msg *msg)
  1811. {
  1812. struct ceph_mds_client *mdsc = session->s_mdsc;
  1813. u32 op;
  1814. u64 seq;
  1815. int mds = session->s_mds;
  1816. struct ceph_mds_session_head *h = msg->front.iov_base;
  1817. int wake = 0;
  1818. /* decode */
  1819. if (msg->front.iov_len != sizeof(*h))
  1820. goto bad;
  1821. op = le32_to_cpu(h->op);
  1822. seq = le64_to_cpu(h->seq);
  1823. mutex_lock(&mdsc->mutex);
  1824. if (op == CEPH_SESSION_CLOSE)
  1825. __unregister_session(mdsc, session);
  1826. /* FIXME: this ttl calculation is generous */
  1827. session->s_ttl = jiffies + HZ*mdsc->mdsmap->m_session_autoclose;
  1828. mutex_unlock(&mdsc->mutex);
  1829. mutex_lock(&session->s_mutex);
  1830. dout("handle_session mds%d %s %p state %s seq %llu\n",
  1831. mds, ceph_session_op_name(op), session,
  1832. session_state_name(session->s_state), seq);
  1833. if (session->s_state == CEPH_MDS_SESSION_HUNG) {
  1834. session->s_state = CEPH_MDS_SESSION_OPEN;
  1835. pr_info("mds%d came back\n", session->s_mds);
  1836. }
  1837. switch (op) {
  1838. case CEPH_SESSION_OPEN:
  1839. if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
  1840. pr_info("mds%d reconnect success\n", session->s_mds);
  1841. session->s_state = CEPH_MDS_SESSION_OPEN;
  1842. renewed_caps(mdsc, session, 0);
  1843. wake = 1;
  1844. if (mdsc->stopping)
  1845. __close_session(mdsc, session);
  1846. break;
  1847. case CEPH_SESSION_RENEWCAPS:
  1848. if (session->s_renew_seq == seq)
  1849. renewed_caps(mdsc, session, 1);
  1850. break;
  1851. case CEPH_SESSION_CLOSE:
  1852. if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
  1853. pr_info("mds%d reconnect denied\n", session->s_mds);
  1854. remove_session_caps(session);
  1855. wake = 1; /* for good measure */
  1856. complete(&mdsc->session_close_waiters);
  1857. kick_requests(mdsc, mds);
  1858. break;
  1859. case CEPH_SESSION_STALE:
  1860. pr_info("mds%d caps went stale, renewing\n",
  1861. session->s_mds);
  1862. spin_lock(&session->s_cap_lock);
  1863. session->s_cap_gen++;
  1864. session->s_cap_ttl = 0;
  1865. spin_unlock(&session->s_cap_lock);
  1866. send_renew_caps(mdsc, session);
  1867. break;
  1868. case CEPH_SESSION_RECALL_STATE:
  1869. trim_caps(mdsc, session, le32_to_cpu(h->max_caps));
  1870. break;
  1871. default:
  1872. pr_err("mdsc_handle_session bad op %d mds%d\n", op, mds);
  1873. WARN_ON(1);
  1874. }
  1875. mutex_unlock(&session->s_mutex);
  1876. if (wake) {
  1877. mutex_lock(&mdsc->mutex);
  1878. __wake_requests(mdsc, &session->s_waiting);
  1879. mutex_unlock(&mdsc->mutex);
  1880. }
  1881. return;
  1882. bad:
  1883. pr_err("mdsc_handle_session corrupt message mds%d len %d\n", mds,
  1884. (int)msg->front.iov_len);
  1885. ceph_msg_dump(msg);
  1886. return;
  1887. }
  1888. /*
  1889. * called under session->mutex.
  1890. */
  1891. static void replay_unsafe_requests(struct ceph_mds_client *mdsc,
  1892. struct ceph_mds_session *session)
  1893. {
  1894. struct ceph_mds_request *req, *nreq;
  1895. int err;
  1896. dout("replay_unsafe_requests mds%d\n", session->s_mds);
  1897. mutex_lock(&mdsc->mutex);
  1898. list_for_each_entry_safe(req, nreq, &session->s_unsafe, r_unsafe_item) {
  1899. err = __prepare_send_request(mdsc, req, session->s_mds);
  1900. if (!err) {
  1901. ceph_msg_get(req->r_request);
  1902. ceph_con_send(&session->s_con, req->r_request);
  1903. }
  1904. }
  1905. mutex_unlock(&mdsc->mutex);
  1906. }
  1907. /*
  1908. * Encode information about a cap for a reconnect with the MDS.
  1909. */
  1910. static int encode_caps_cb(struct inode *inode, struct ceph_cap *cap,
  1911. void *arg)
  1912. {
  1913. struct ceph_mds_cap_reconnect rec;
  1914. struct ceph_inode_info *ci;
  1915. struct ceph_pagelist *pagelist = arg;
  1916. char *path;
  1917. int pathlen, err;
  1918. u64 pathbase;
  1919. struct dentry *dentry;
  1920. ci = cap->ci;
  1921. dout(" adding %p ino %llx.%llx cap %p %lld %s\n",
  1922. inode, ceph_vinop(inode), cap, cap->cap_id,
  1923. ceph_cap_string(cap->issued));
  1924. err = ceph_pagelist_encode_64(pagelist, ceph_ino(inode));
  1925. if (err)
  1926. return err;
  1927. dentry = d_find_alias(inode);
  1928. if (dentry) {
  1929. path = ceph_mdsc_build_path(dentry, &pathlen, &pathbase, 0);
  1930. if (IS_ERR(path)) {
  1931. err = PTR_ERR(path);
  1932. BUG_ON(err);
  1933. }
  1934. } else {
  1935. path = NULL;
  1936. pathlen = 0;
  1937. }
  1938. err = ceph_pagelist_encode_string(pagelist, path, pathlen);
  1939. if (err)
  1940. goto out;
  1941. spin_lock(&inode->i_lock);
  1942. cap->seq = 0; /* reset cap seq */
  1943. cap->issue_seq = 0; /* and issue_seq */
  1944. rec.cap_id = cpu_to_le64(cap->cap_id);
  1945. rec.pathbase = cpu_to_le64(pathbase);
  1946. rec.wanted = cpu_to_le32(__ceph_caps_wanted(ci));
  1947. rec.issued = cpu_to_le32(cap->issued);
  1948. rec.size = cpu_to_le64(inode->i_size);
  1949. ceph_encode_timespec(&rec.mtime, &inode->i_mtime);
  1950. ceph_encode_timespec(&rec.atime, &inode->i_atime);
  1951. rec.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);
  1952. spin_unlock(&inode->i_lock);
  1953. err = ceph_pagelist_append(pagelist, &rec, sizeof(rec));
  1954. out:
  1955. kfree(path);
  1956. dput(dentry);
  1957. return err;
  1958. }
  1959. /*
  1960. * If an MDS fails and recovers, clients need to reconnect in order to
  1961. * reestablish shared state. This includes all caps issued through
  1962. * this session _and_ the snap_realm hierarchy. Because it's not
  1963. * clear which snap realms the mds cares about, we send everything we
  1964. * know about.. that ensures we'll then get any new info the
  1965. * recovering MDS might have.
  1966. *
  1967. * This is a relatively heavyweight operation, but it's rare.
  1968. *
  1969. * called with mdsc->mutex held.
  1970. */
  1971. static void send_mds_reconnect(struct ceph_mds_client *mdsc,
  1972. struct ceph_mds_session *session)
  1973. {
  1974. struct ceph_msg *reply;
  1975. struct rb_node *p;
  1976. int mds = session->s_mds;
  1977. int err = -ENOMEM;
  1978. struct ceph_pagelist *pagelist;
  1979. pr_info("mds%d reconnect start\n", mds);
  1980. pagelist = kmalloc(sizeof(*pagelist), GFP_NOFS);
  1981. if (!pagelist)
  1982. goto fail_nopagelist;
  1983. ceph_pagelist_init(pagelist);
  1984. reply = ceph_msg_new(CEPH_MSG_CLIENT_RECONNECT, 0, GFP_NOFS);
  1985. if (!reply)
  1986. goto fail_nomsg;
  1987. mutex_lock(&session->s_mutex);
  1988. session->s_state = CEPH_MDS_SESSION_RECONNECTING;
  1989. session->s_seq = 0;
  1990. ceph_con_open(&session->s_con,
  1991. ceph_mdsmap_get_addr(mdsc->mdsmap, mds));
  1992. /* replay unsafe requests */
  1993. replay_unsafe_requests(mdsc, session);
  1994. down_read(&mdsc->snap_rwsem);
  1995. dout("session %p state %s\n", session,
  1996. session_state_name(session->s_state));
  1997. /* drop old cap expires; we're about to reestablish that state */
  1998. discard_cap_releases(mdsc, session);
  1999. /* traverse this session's caps */
  2000. err = ceph_pagelist_encode_32(pagelist, session->s_nr_caps);
  2001. if (err)
  2002. goto fail;
  2003. err = iterate_session_caps(session, encode_caps_cb, pagelist);
  2004. if (err < 0)
  2005. goto fail;
  2006. /*
  2007. * snaprealms. we provide mds with the ino, seq (version), and
  2008. * parent for all of our realms. If the mds has any newer info,
  2009. * it will tell us.
  2010. */
  2011. for (p = rb_first(&mdsc->snap_realms); p; p = rb_next(p)) {
  2012. struct ceph_snap_realm *realm =
  2013. rb_entry(p, struct ceph_snap_realm, node);
  2014. struct ceph_mds_snaprealm_reconnect sr_rec;
  2015. dout(" adding snap realm %llx seq %lld parent %llx\n",
  2016. realm->ino, realm->seq, realm->parent_ino);
  2017. sr_rec.ino = cpu_to_le64(realm->ino);
  2018. sr_rec.seq = cpu_to_le64(realm->seq);
  2019. sr_rec.parent = cpu_to_le64(realm->parent_ino);
  2020. err = ceph_pagelist_append(pagelist, &sr_rec, sizeof(sr_rec));
  2021. if (err)
  2022. goto fail;
  2023. }
  2024. reply->pagelist = pagelist;
  2025. reply->hdr.data_len = cpu_to_le32(pagelist->length);
  2026. reply->nr_pages = calc_pages_for(0, pagelist->length);
  2027. ceph_con_send(&session->s_con, reply);
  2028. mutex_unlock(&session->s_mutex);
  2029. mutex_lock(&mdsc->mutex);
  2030. __wake_requests(mdsc, &session->s_waiting);
  2031. mutex_unlock(&mdsc->mutex);
  2032. up_read(&mdsc->snap_rwsem);
  2033. return;
  2034. fail:
  2035. ceph_msg_put(reply);
  2036. up_read(&mdsc->snap_rwsem);
  2037. mutex_unlock(&session->s_mutex);
  2038. fail_nomsg:
  2039. ceph_pagelist_release(pagelist);
  2040. kfree(pagelist);
  2041. fail_nopagelist:
  2042. pr_err("error %d preparing reconnect for mds%d\n", err, mds);
  2043. return;
  2044. }
  2045. /*
  2046. * compare old and new mdsmaps, kicking requests
  2047. * and closing out old connections as necessary
  2048. *
  2049. * called under mdsc->mutex.
  2050. */
  2051. static void check_new_map(struct ceph_mds_client *mdsc,
  2052. struct ceph_mdsmap *newmap,
  2053. struct ceph_mdsmap *oldmap)
  2054. {
  2055. int i;
  2056. int oldstate, newstate;
  2057. struct ceph_mds_session *s;
  2058. dout("check_new_map new %u old %u\n",
  2059. newmap->m_epoch, oldmap->m_epoch);
  2060. for (i = 0; i < oldmap->m_max_mds && i < mdsc->max_sessions; i++) {
  2061. if (mdsc->sessions[i] == NULL)
  2062. continue;
  2063. s = mdsc->sessions[i];
  2064. oldstate = ceph_mdsmap_get_state(oldmap, i);
  2065. newstate = ceph_mdsmap_get_state(newmap, i);
  2066. dout("check_new_map mds%d state %s -> %s (session %s)\n",
  2067. i, ceph_mds_state_name(oldstate),
  2068. ceph_mds_state_name(newstate),
  2069. session_state_name(s->s_state));
  2070. if (memcmp(ceph_mdsmap_get_addr(oldmap, i),
  2071. ceph_mdsmap_get_addr(newmap, i),
  2072. sizeof(struct ceph_entity_addr))) {
  2073. if (s->s_state == CEPH_MDS_SESSION_OPENING) {
  2074. /* the session never opened, just close it
  2075. * out now */
  2076. __wake_requests(mdsc, &s->s_waiting);
  2077. __unregister_session(mdsc, s);
  2078. } else {
  2079. /* just close it */
  2080. mutex_unlock(&mdsc->mutex);
  2081. mutex_lock(&s->s_mutex);
  2082. mutex_lock(&mdsc->mutex);
  2083. ceph_con_close(&s->s_con);
  2084. mutex_unlock(&s->s_mutex);
  2085. s->s_state = CEPH_MDS_SESSION_RESTARTING;
  2086. }
  2087. /* kick any requests waiting on the recovering mds */
  2088. kick_requests(mdsc, i);
  2089. } else if (oldstate == newstate) {
  2090. continue; /* nothing new with this mds */
  2091. }
  2092. /*
  2093. * send reconnect?
  2094. */
  2095. if (s->s_state == CEPH_MDS_SESSION_RESTARTING &&
  2096. newstate >= CEPH_MDS_STATE_RECONNECT) {
  2097. mutex_unlock(&mdsc->mutex);
  2098. send_mds_reconnect(mdsc, s);
  2099. mutex_lock(&mdsc->mutex);
  2100. }
  2101. /*
  2102. * kick request on any mds that has gone active.
  2103. */
  2104. if (oldstate < CEPH_MDS_STATE_ACTIVE &&
  2105. newstate >= CEPH_MDS_STATE_ACTIVE) {
  2106. if (oldstate != CEPH_MDS_STATE_CREATING &&
  2107. oldstate != CEPH_MDS_STATE_STARTING)
  2108. pr_info("mds%d recovery completed\n", s->s_mds);
  2109. kick_requests(mdsc, i);
  2110. ceph_kick_flushing_caps(mdsc, s);
  2111. wake_up_session_caps(s, 1);
  2112. }
  2113. }
  2114. }
  2115. /*
  2116. * leases
  2117. */
  2118. /*
  2119. * caller must hold session s_mutex, dentry->d_lock
  2120. */
  2121. void __ceph_mdsc_drop_dentry_lease(struct dentry *dentry)
  2122. {
  2123. struct ceph_dentry_info *di = ceph_dentry(dentry);
  2124. ceph_put_mds_session(di->lease_session);
  2125. di->lease_session = NULL;
  2126. }
  2127. static void handle_lease(struct ceph_mds_client *mdsc,
  2128. struct ceph_mds_session *session,
  2129. struct ceph_msg *msg)
  2130. {
  2131. struct super_block *sb = mdsc->client->sb;
  2132. struct inode *inode;
  2133. struct ceph_inode_info *ci;
  2134. struct dentry *parent, *dentry;
  2135. struct ceph_dentry_info *di;
  2136. int mds = session->s_mds;
  2137. struct ceph_mds_lease *h = msg->front.iov_base;
  2138. struct ceph_vino vino;
  2139. int mask;
  2140. struct qstr dname;
  2141. int release = 0;
  2142. dout("handle_lease from mds%d\n", mds);
  2143. /* decode */
  2144. if (msg->front.iov_len < sizeof(*h) + sizeof(u32))
  2145. goto bad;
  2146. vino.ino = le64_to_cpu(h->ino);
  2147. vino.snap = CEPH_NOSNAP;
  2148. mask = le16_to_cpu(h->mask);
  2149. dname.name = (void *)h + sizeof(*h) + sizeof(u32);
  2150. dname.len = msg->front.iov_len - sizeof(*h) - sizeof(u32);
  2151. if (dname.len != get_unaligned_le32(h+1))
  2152. goto bad;
  2153. mutex_lock(&session->s_mutex);
  2154. session->s_seq++;
  2155. /* lookup inode */
  2156. inode = ceph_find_inode(sb, vino);
  2157. dout("handle_lease '%s', mask %d, ino %llx %p\n",
  2158. ceph_lease_op_name(h->action), mask, vino.ino, inode);
  2159. if (inode == NULL) {
  2160. dout("handle_lease no inode %llx\n", vino.ino);
  2161. goto release;
  2162. }
  2163. ci = ceph_inode(inode);
  2164. /* dentry */
  2165. parent = d_find_alias(inode);
  2166. if (!parent) {
  2167. dout("no parent dentry on inode %p\n", inode);
  2168. WARN_ON(1);
  2169. goto release; /* hrm... */
  2170. }
  2171. dname.hash = full_name_hash(dname.name, dname.len);
  2172. dentry = d_lookup(parent, &dname);
  2173. dput(parent);
  2174. if (!dentry)
  2175. goto release;
  2176. spin_lock(&dentry->d_lock);
  2177. di = ceph_dentry(dentry);
  2178. switch (h->action) {
  2179. case CEPH_MDS_LEASE_REVOKE:
  2180. if (di && di->lease_session == session) {
  2181. h->seq = cpu_to_le32(di->lease_seq);
  2182. __ceph_mdsc_drop_dentry_lease(dentry);
  2183. }
  2184. release = 1;
  2185. break;
  2186. case CEPH_MDS_LEASE_RENEW:
  2187. if (di && di->lease_session == session &&
  2188. di->lease_gen == session->s_cap_gen &&
  2189. di->lease_renew_from &&
  2190. di->lease_renew_after == 0) {
  2191. unsigned long duration =
  2192. le32_to_cpu(h->duration_ms) * HZ / 1000;
  2193. di->lease_seq = le32_to_cpu(h->seq);
  2194. dentry->d_time = di->lease_renew_from + duration;
  2195. di->lease_renew_after = di->lease_renew_from +
  2196. (duration >> 1);
  2197. di->lease_renew_from = 0;
  2198. }
  2199. break;
  2200. }
  2201. spin_unlock(&dentry->d_lock);
  2202. dput(dentry);
  2203. if (!release)
  2204. goto out;
  2205. release:
  2206. /* let's just reuse the same message */
  2207. h->action = CEPH_MDS_LEASE_REVOKE_ACK;
  2208. ceph_msg_get(msg);
  2209. ceph_con_send(&session->s_con, msg);
  2210. out:
  2211. iput(inode);
  2212. mutex_unlock(&session->s_mutex);
  2213. return;
  2214. bad:
  2215. pr_err("corrupt lease message\n");
  2216. ceph_msg_dump(msg);
  2217. }
  2218. void ceph_mdsc_lease_send_msg(struct ceph_mds_session *session,
  2219. struct inode *inode,
  2220. struct dentry *dentry, char action,
  2221. u32 seq)
  2222. {
  2223. struct ceph_msg *msg;
  2224. struct ceph_mds_lease *lease;
  2225. int len = sizeof(*lease) + sizeof(u32);
  2226. int dnamelen = 0;
  2227. dout("lease_send_msg inode %p dentry %p %s to mds%d\n",
  2228. inode, dentry, ceph_lease_op_name(action), session->s_mds);
  2229. dnamelen = dentry->d_name.len;
  2230. len += dnamelen;
  2231. msg = ceph_msg_new(CEPH_MSG_CLIENT_LEASE, len, GFP_NOFS);
  2232. if (!msg)
  2233. return;
  2234. lease = msg->front.iov_base;
  2235. lease->action = action;
  2236. lease->mask = cpu_to_le16(CEPH_LOCK_DN);
  2237. lease->ino = cpu_to_le64(ceph_vino(inode).ino);
  2238. lease->first = lease->last = cpu_to_le64(ceph_vino(inode).snap);
  2239. lease->seq = cpu_to_le32(seq);
  2240. put_unaligned_le32(dnamelen, lease + 1);
  2241. memcpy((void *)(lease + 1) + 4, dentry->d_name.name, dnamelen);
  2242. /*
  2243. * if this is a preemptive lease RELEASE, no need to
  2244. * flush request stream, since the actual request will
  2245. * soon follow.
  2246. */
  2247. msg->more_to_follow = (action == CEPH_MDS_LEASE_RELEASE);
  2248. ceph_con_send(&session->s_con, msg);
  2249. }
  2250. /*
  2251. * Preemptively release a lease we expect to invalidate anyway.
  2252. * Pass @inode always, @dentry is optional.
  2253. */
  2254. void ceph_mdsc_lease_release(struct ceph_mds_client *mdsc, struct inode *inode,
  2255. struct dentry *dentry, int mask)
  2256. {
  2257. struct ceph_dentry_info *di;
  2258. struct ceph_mds_session *session;
  2259. u32 seq;
  2260. BUG_ON(inode == NULL);
  2261. BUG_ON(dentry == NULL);
  2262. BUG_ON(mask != CEPH_LOCK_DN);
  2263. /* is dentry lease valid? */
  2264. spin_lock(&dentry->d_lock);
  2265. di = ceph_dentry(dentry);
  2266. if (!di || !di->lease_session ||
  2267. di->lease_session->s_mds < 0 ||
  2268. di->lease_gen != di->lease_session->s_cap_gen ||
  2269. !time_before(jiffies, dentry->d_time)) {
  2270. dout("lease_release inode %p dentry %p -- "
  2271. "no lease on %d\n",
  2272. inode, dentry, mask);
  2273. spin_unlock(&dentry->d_lock);
  2274. return;
  2275. }
  2276. /* we do have a lease on this dentry; note mds and seq */
  2277. session = ceph_get_mds_session(di->lease_session);
  2278. seq = di->lease_seq;
  2279. __ceph_mdsc_drop_dentry_lease(dentry);
  2280. spin_unlock(&dentry->d_lock);
  2281. dout("lease_release inode %p dentry %p mask %d to mds%d\n",
  2282. inode, dentry, mask, session->s_mds);
  2283. ceph_mdsc_lease_send_msg(session, inode, dentry,
  2284. CEPH_MDS_LEASE_RELEASE, seq);
  2285. ceph_put_mds_session(session);
  2286. }
  2287. /*
  2288. * drop all leases (and dentry refs) in preparation for umount
  2289. */
  2290. static void drop_leases(struct ceph_mds_client *mdsc)
  2291. {
  2292. int i;
  2293. dout("drop_leases\n");
  2294. mutex_lock(&mdsc->mutex);
  2295. for (i = 0; i < mdsc->max_sessions; i++) {
  2296. struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);
  2297. if (!s)
  2298. continue;
  2299. mutex_unlock(&mdsc->mutex);
  2300. mutex_lock(&s->s_mutex);
  2301. mutex_unlock(&s->s_mutex);
  2302. ceph_put_mds_session(s);
  2303. mutex_lock(&mdsc->mutex);
  2304. }
  2305. mutex_unlock(&mdsc->mutex);
  2306. }
  2307. /*
  2308. * delayed work -- periodically trim expired leases, renew caps with mds
  2309. */
  2310. static void schedule_delayed(struct ceph_mds_client *mdsc)
  2311. {
  2312. int delay = 5;
  2313. unsigned hz = round_jiffies_relative(HZ * delay);
  2314. schedule_delayed_work(&mdsc->delayed_work, hz);
  2315. }
  2316. static void delayed_work(struct work_struct *work)
  2317. {
  2318. int i;
  2319. struct ceph_mds_client *mdsc =
  2320. container_of(work, struct ceph_mds_client, delayed_work.work);
  2321. int renew_interval;
  2322. int renew_caps;
  2323. dout("mdsc delayed_work\n");
  2324. ceph_check_delayed_caps(mdsc);
  2325. mutex_lock(&mdsc->mutex);
  2326. renew_interval = mdsc->mdsmap->m_session_timeout >> 2;
  2327. renew_caps = time_after_eq(jiffies, HZ*renew_interval +
  2328. mdsc->last_renew_caps);
  2329. if (renew_caps)
  2330. mdsc->last_renew_caps = jiffies;
  2331. for (i = 0; i < mdsc->max_sessions; i++) {
  2332. struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);
  2333. if (s == NULL)
  2334. continue;
  2335. if (s->s_state == CEPH_MDS_SESSION_CLOSING) {
  2336. dout("resending session close request for mds%d\n",
  2337. s->s_mds);
  2338. request_close_session(mdsc, s);
  2339. ceph_put_mds_session(s);
  2340. continue;
  2341. }
  2342. if (s->s_ttl && time_after(jiffies, s->s_ttl)) {
  2343. if (s->s_state == CEPH_MDS_SESSION_OPEN) {
  2344. s->s_state = CEPH_MDS_SESSION_HUNG;
  2345. pr_info("mds%d hung\n", s->s_mds);
  2346. }
  2347. }
  2348. if (s->s_state < CEPH_MDS_SESSION_OPEN) {
  2349. /* this mds is failed or recovering, just wait */
  2350. ceph_put_mds_session(s);
  2351. continue;
  2352. }
  2353. mutex_unlock(&mdsc->mutex);
  2354. mutex_lock(&s->s_mutex);
  2355. if (renew_caps)
  2356. send_renew_caps(mdsc, s);
  2357. else
  2358. ceph_con_keepalive(&s->s_con);
  2359. add_cap_releases(mdsc, s, -1);
  2360. if (s->s_state == CEPH_MDS_SESSION_OPEN ||
  2361. s->s_state == CEPH_MDS_SESSION_HUNG)
  2362. send_cap_releases(mdsc, s);
  2363. mutex_unlock(&s->s_mutex);
  2364. ceph_put_mds_session(s);
  2365. mutex_lock(&mdsc->mutex);
  2366. }
  2367. mutex_unlock(&mdsc->mutex);
  2368. schedule_delayed(mdsc);
  2369. }
  2370. int ceph_mdsc_init(struct ceph_mds_client *mdsc, struct ceph_client *client)
  2371. {
  2372. mdsc->client = client;
  2373. mutex_init(&mdsc->mutex);
  2374. mdsc->mdsmap = kzalloc(sizeof(*mdsc->mdsmap), GFP_NOFS);
  2375. if (mdsc->mdsmap == NULL)
  2376. return -ENOMEM;
  2377. init_completion(&mdsc->safe_umount_waiters);
  2378. init_completion(&mdsc->session_close_waiters);
  2379. INIT_LIST_HEAD(&mdsc->waiting_for_map);
  2380. mdsc->sessions = NULL;
  2381. mdsc->max_sessions = 0;
  2382. mdsc->stopping = 0;
  2383. init_rwsem(&mdsc->snap_rwsem);
  2384. mdsc->snap_realms = RB_ROOT;
  2385. INIT_LIST_HEAD(&mdsc->snap_empty);
  2386. spin_lock_init(&mdsc->snap_empty_lock);
  2387. mdsc->last_tid = 0;
  2388. mdsc->request_tree = RB_ROOT;
  2389. INIT_DELAYED_WORK(&mdsc->delayed_work, delayed_work);
  2390. mdsc->last_renew_caps = jiffies;
  2391. INIT_LIST_HEAD(&mdsc->cap_delay_list);
  2392. spin_lock_init(&mdsc->cap_delay_lock);
  2393. INIT_LIST_HEAD(&mdsc->snap_flush_list);
  2394. spin_lock_init(&mdsc->snap_flush_lock);
  2395. mdsc->cap_flush_seq = 0;
  2396. INIT_LIST_HEAD(&mdsc->cap_dirty);
  2397. mdsc->num_cap_flushing = 0;
  2398. spin_lock_init(&mdsc->cap_dirty_lock);
  2399. init_waitqueue_head(&mdsc->cap_flushing_wq);
  2400. spin_lock_init(&mdsc->dentry_lru_lock);
  2401. INIT_LIST_HEAD(&mdsc->dentry_lru);
  2402. return 0;
  2403. }
  2404. /*
  2405. * Wait for safe replies on open mds requests. If we time out, drop
  2406. * all requests from the tree to avoid dangling dentry refs.
  2407. */
  2408. static void wait_requests(struct ceph_mds_client *mdsc)
  2409. {
  2410. struct ceph_mds_request *req;
  2411. struct ceph_client *client = mdsc->client;
  2412. mutex_lock(&mdsc->mutex);
  2413. if (__get_oldest_req(mdsc)) {
  2414. mutex_unlock(&mdsc->mutex);
  2415. dout("wait_requests waiting for requests\n");
  2416. wait_for_completion_timeout(&mdsc->safe_umount_waiters,
  2417. client->mount_args->mount_timeout * HZ);
  2418. /* tear down remaining requests */
  2419. mutex_lock(&mdsc->mutex);
  2420. while ((req = __get_oldest_req(mdsc))) {
  2421. dout("wait_requests timed out on tid %llu\n",
  2422. req->r_tid);
  2423. __unregister_request(mdsc, req);
  2424. }
  2425. }
  2426. mutex_unlock(&mdsc->mutex);
  2427. dout("wait_requests done\n");
  2428. }
  2429. /*
  2430. * called before mount is ro, and before dentries are torn down.
  2431. * (hmm, does this still race with new lookups?)
  2432. */
  2433. void ceph_mdsc_pre_umount(struct ceph_mds_client *mdsc)
  2434. {
  2435. dout("pre_umount\n");
  2436. mdsc->stopping = 1;
  2437. drop_leases(mdsc);
  2438. ceph_flush_dirty_caps(mdsc);
  2439. wait_requests(mdsc);
  2440. }
  2441. /*
  2442. * wait for all write mds requests to flush.
  2443. */
  2444. static void wait_unsafe_requests(struct ceph_mds_client *mdsc, u64 want_tid)
  2445. {
  2446. struct ceph_mds_request *req = NULL, *nextreq;
  2447. struct rb_node *n;
  2448. mutex_lock(&mdsc->mutex);
  2449. dout("wait_unsafe_requests want %lld\n", want_tid);
  2450. restart:
  2451. req = __get_oldest_req(mdsc);
  2452. while (req && req->r_tid <= want_tid) {
  2453. /* find next request */
  2454. n = rb_next(&req->r_node);
  2455. if (n)
  2456. nextreq = rb_entry(n, struct ceph_mds_request, r_node);
  2457. else
  2458. nextreq = NULL;
  2459. if ((req->r_op & CEPH_MDS_OP_WRITE)) {
  2460. /* write op */
  2461. ceph_mdsc_get_request(req);
  2462. if (nextreq)
  2463. ceph_mdsc_get_request(nextreq);
  2464. mutex_unlock(&mdsc->mutex);
  2465. dout("wait_unsafe_requests wait on %llu (want %llu)\n",
  2466. req->r_tid, want_tid);
  2467. wait_for_completion(&req->r_safe_completion);
  2468. mutex_lock(&mdsc->mutex);
  2469. ceph_mdsc_put_request(req);
  2470. if (!nextreq)
  2471. break; /* next dne before, so we're done! */
  2472. if (RB_EMPTY_NODE(&nextreq->r_node)) {
  2473. /* next request was removed from tree */
  2474. ceph_mdsc_put_request(nextreq);
  2475. goto restart;
  2476. }
  2477. ceph_mdsc_put_request(nextreq); /* won't go away */
  2478. }
  2479. req = nextreq;
  2480. }
  2481. mutex_unlock(&mdsc->mutex);
  2482. dout("wait_unsafe_requests done\n");
  2483. }
  2484. void ceph_mdsc_sync(struct ceph_mds_client *mdsc)
  2485. {
  2486. u64 want_tid, want_flush;
  2487. if (mdsc->client->mount_state == CEPH_MOUNT_SHUTDOWN)
  2488. return;
  2489. dout("sync\n");
  2490. mutex_lock(&mdsc->mutex);
  2491. want_tid = mdsc->last_tid;
  2492. want_flush = mdsc->cap_flush_seq;
  2493. mutex_unlock(&mdsc->mutex);
  2494. dout("sync want tid %lld flush_seq %lld\n", want_tid, want_flush);
  2495. ceph_flush_dirty_caps(mdsc);
  2496. wait_unsafe_requests(mdsc, want_tid);
  2497. wait_event(mdsc->cap_flushing_wq, check_cap_flush(mdsc, want_flush));
  2498. }
  2499. /*
  2500. * called after sb is ro.
  2501. */
  2502. void ceph_mdsc_close_sessions(struct ceph_mds_client *mdsc)
  2503. {
  2504. struct ceph_mds_session *session;
  2505. int i;
  2506. int n;
  2507. struct ceph_client *client = mdsc->client;
  2508. unsigned long started, timeout = client->mount_args->mount_timeout * HZ;
  2509. dout("close_sessions\n");
  2510. mutex_lock(&mdsc->mutex);
  2511. /* close sessions */
  2512. started = jiffies;
  2513. while (time_before(jiffies, started + timeout)) {
  2514. dout("closing sessions\n");
  2515. n = 0;
  2516. for (i = 0; i < mdsc->max_sessions; i++) {
  2517. session = __ceph_lookup_mds_session(mdsc, i);
  2518. if (!session)
  2519. continue;
  2520. mutex_unlock(&mdsc->mutex);
  2521. mutex_lock(&session->s_mutex);
  2522. __close_session(mdsc, session);
  2523. mutex_unlock(&session->s_mutex);
  2524. ceph_put_mds_session(session);
  2525. mutex_lock(&mdsc->mutex);
  2526. n++;
  2527. }
  2528. if (n == 0)
  2529. break;
  2530. if (client->mount_state == CEPH_MOUNT_SHUTDOWN)
  2531. break;
  2532. dout("waiting for sessions to close\n");
  2533. mutex_unlock(&mdsc->mutex);
  2534. wait_for_completion_timeout(&mdsc->session_close_waiters,
  2535. timeout);
  2536. mutex_lock(&mdsc->mutex);
  2537. }
  2538. /* tear down remaining sessions */
  2539. for (i = 0; i < mdsc->max_sessions; i++) {
  2540. if (mdsc->sessions[i]) {
  2541. session = get_session(mdsc->sessions[i]);
  2542. __unregister_session(mdsc, session);
  2543. mutex_unlock(&mdsc->mutex);
  2544. mutex_lock(&session->s_mutex);
  2545. remove_session_caps(session);
  2546. mutex_unlock(&session->s_mutex);
  2547. ceph_put_mds_session(session);
  2548. mutex_lock(&mdsc->mutex);
  2549. }
  2550. }
  2551. WARN_ON(!list_empty(&mdsc->cap_delay_list));
  2552. mutex_unlock(&mdsc->mutex);
  2553. ceph_cleanup_empty_realms(mdsc);
  2554. cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */
  2555. dout("stopped\n");
  2556. }
  2557. void ceph_mdsc_stop(struct ceph_mds_client *mdsc)
  2558. {
  2559. dout("stop\n");
  2560. cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */
  2561. if (mdsc->mdsmap)
  2562. ceph_mdsmap_destroy(mdsc->mdsmap);
  2563. kfree(mdsc->sessions);
  2564. }
  2565. /*
  2566. * handle mds map update.
  2567. */
  2568. void ceph_mdsc_handle_map(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
  2569. {
  2570. u32 epoch;
  2571. u32 maplen;
  2572. void *p = msg->front.iov_base;
  2573. void *end = p + msg->front.iov_len;
  2574. struct ceph_mdsmap *newmap, *oldmap;
  2575. struct ceph_fsid fsid;
  2576. int err = -EINVAL;
  2577. ceph_decode_need(&p, end, sizeof(fsid)+2*sizeof(u32), bad);
  2578. ceph_decode_copy(&p, &fsid, sizeof(fsid));
  2579. if (ceph_check_fsid(mdsc->client, &fsid) < 0)
  2580. return;
  2581. epoch = ceph_decode_32(&p);
  2582. maplen = ceph_decode_32(&p);
  2583. dout("handle_map epoch %u len %d\n", epoch, (int)maplen);
  2584. /* do we need it? */
  2585. ceph_monc_got_mdsmap(&mdsc->client->monc, epoch);
  2586. mutex_lock(&mdsc->mutex);
  2587. if (mdsc->mdsmap && epoch <= mdsc->mdsmap->m_epoch) {
  2588. dout("handle_map epoch %u <= our %u\n",
  2589. epoch, mdsc->mdsmap->m_epoch);
  2590. mutex_unlock(&mdsc->mutex);
  2591. return;
  2592. }
  2593. newmap = ceph_mdsmap_decode(&p, end);
  2594. if (IS_ERR(newmap)) {
  2595. err = PTR_ERR(newmap);
  2596. goto bad_unlock;
  2597. }
  2598. /* swap into place */
  2599. if (mdsc->mdsmap) {
  2600. oldmap = mdsc->mdsmap;
  2601. mdsc->mdsmap = newmap;
  2602. check_new_map(mdsc, newmap, oldmap);
  2603. ceph_mdsmap_destroy(oldmap);
  2604. } else {
  2605. mdsc->mdsmap = newmap; /* first mds map */
  2606. }
  2607. mdsc->client->sb->s_maxbytes = mdsc->mdsmap->m_max_file_size;
  2608. __wake_requests(mdsc, &mdsc->waiting_for_map);
  2609. mutex_unlock(&mdsc->mutex);
  2610. schedule_delayed(mdsc);
  2611. return;
  2612. bad_unlock:
  2613. mutex_unlock(&mdsc->mutex);
  2614. bad:
  2615. pr_err("error decoding mdsmap %d\n", err);
  2616. return;
  2617. }
  2618. static struct ceph_connection *con_get(struct ceph_connection *con)
  2619. {
  2620. struct ceph_mds_session *s = con->private;
  2621. if (get_session(s)) {
  2622. dout("mdsc con_get %p ok (%d)\n", s, atomic_read(&s->s_ref));
  2623. return con;
  2624. }
  2625. dout("mdsc con_get %p FAIL\n", s);
  2626. return NULL;
  2627. }
  2628. static void con_put(struct ceph_connection *con)
  2629. {
  2630. struct ceph_mds_session *s = con->private;
  2631. ceph_put_mds_session(s);
  2632. dout("mdsc con_put %p (%d)\n", s, atomic_read(&s->s_ref));
  2633. }
  2634. /*
  2635. * if the client is unresponsive for long enough, the mds will kill
  2636. * the session entirely.
  2637. */
  2638. static void peer_reset(struct ceph_connection *con)
  2639. {
  2640. struct ceph_mds_session *s = con->private;
  2641. struct ceph_mds_client *mdsc = s->s_mdsc;
  2642. pr_warning("mds%d closed our session\n", s->s_mds);
  2643. send_mds_reconnect(mdsc, s);
  2644. }
  2645. static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
  2646. {
  2647. struct ceph_mds_session *s = con->private;
  2648. struct ceph_mds_client *mdsc = s->s_mdsc;
  2649. int type = le16_to_cpu(msg->hdr.type);
  2650. mutex_lock(&mdsc->mutex);
  2651. if (__verify_registered_session(mdsc, s) < 0) {
  2652. mutex_unlock(&mdsc->mutex);
  2653. goto out;
  2654. }
  2655. mutex_unlock(&mdsc->mutex);
  2656. switch (type) {
  2657. case CEPH_MSG_MDS_MAP:
  2658. ceph_mdsc_handle_map(mdsc, msg);
  2659. break;
  2660. case CEPH_MSG_CLIENT_SESSION:
  2661. handle_session(s, msg);
  2662. break;
  2663. case CEPH_MSG_CLIENT_REPLY:
  2664. handle_reply(s, msg);
  2665. break;
  2666. case CEPH_MSG_CLIENT_REQUEST_FORWARD:
  2667. handle_forward(mdsc, s, msg);
  2668. break;
  2669. case CEPH_MSG_CLIENT_CAPS:
  2670. ceph_handle_caps(s, msg);
  2671. break;
  2672. case CEPH_MSG_CLIENT_SNAP:
  2673. ceph_handle_snap(mdsc, s, msg);
  2674. break;
  2675. case CEPH_MSG_CLIENT_LEASE:
  2676. handle_lease(mdsc, s, msg);
  2677. break;
  2678. default:
  2679. pr_err("received unknown message type %d %s\n", type,
  2680. ceph_msg_type_name(type));
  2681. }
  2682. out:
  2683. ceph_msg_put(msg);
  2684. }
  2685. /*
  2686. * authentication
  2687. */
  2688. static int get_authorizer(struct ceph_connection *con,
  2689. void **buf, int *len, int *proto,
  2690. void **reply_buf, int *reply_len, int force_new)
  2691. {
  2692. struct ceph_mds_session *s = con->private;
  2693. struct ceph_mds_client *mdsc = s->s_mdsc;
  2694. struct ceph_auth_client *ac = mdsc->client->monc.auth;
  2695. int ret = 0;
  2696. if (force_new && s->s_authorizer) {
  2697. ac->ops->destroy_authorizer(ac, s->s_authorizer);
  2698. s->s_authorizer = NULL;
  2699. }
  2700. if (s->s_authorizer == NULL) {
  2701. if (ac->ops->create_authorizer) {
  2702. ret = ac->ops->create_authorizer(
  2703. ac, CEPH_ENTITY_TYPE_MDS,
  2704. &s->s_authorizer,
  2705. &s->s_authorizer_buf,
  2706. &s->s_authorizer_buf_len,
  2707. &s->s_authorizer_reply_buf,
  2708. &s->s_authorizer_reply_buf_len);
  2709. if (ret)
  2710. return ret;
  2711. }
  2712. }
  2713. *proto = ac->protocol;
  2714. *buf = s->s_authorizer_buf;
  2715. *len = s->s_authorizer_buf_len;
  2716. *reply_buf = s->s_authorizer_reply_buf;
  2717. *reply_len = s->s_authorizer_reply_buf_len;
  2718. return 0;
  2719. }
  2720. static int verify_authorizer_reply(struct ceph_connection *con, int len)
  2721. {
  2722. struct ceph_mds_session *s = con->private;
  2723. struct ceph_mds_client *mdsc = s->s_mdsc;
  2724. struct ceph_auth_client *ac = mdsc->client->monc.auth;
  2725. return ac->ops->verify_authorizer_reply(ac, s->s_authorizer, len);
  2726. }
  2727. static int invalidate_authorizer(struct ceph_connection *con)
  2728. {
  2729. struct ceph_mds_session *s = con->private;
  2730. struct ceph_mds_client *mdsc = s->s_mdsc;
  2731. struct ceph_auth_client *ac = mdsc->client->monc.auth;
  2732. if (ac->ops->invalidate_authorizer)
  2733. ac->ops->invalidate_authorizer(ac, CEPH_ENTITY_TYPE_MDS);
  2734. return ceph_monc_validate_auth(&mdsc->client->monc);
  2735. }
  2736. static const struct ceph_connection_operations mds_con_ops = {
  2737. .get = con_get,
  2738. .put = con_put,
  2739. .dispatch = dispatch,
  2740. .get_authorizer = get_authorizer,
  2741. .verify_authorizer_reply = verify_authorizer_reply,
  2742. .invalidate_authorizer = invalidate_authorizer,
  2743. .peer_reset = peer_reset,
  2744. };
  2745. /* eof */