mds_client.c 133 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875287628772878287928802881288228832884288528862887288828892890289128922893289428952896289728982899290029012902290329042905290629072908290929102911291229132914291529162917291829192920292129222923292429252926292729282929293029312932293329342935293629372938293929402941294229432944294529462947294829492950295129522953295429552956295729582959296029612962296329642965296629672968296929702971297229732974297529762977297829792980298129822983298429852986298729882989299029912992299329942995299629972998299930003001300230033004300530063007300830093010301130123013301430153016301730183019302030213022302330243025302630273028302930303031303230333034303530363037303830393040304130423043304430453046304730483049305030513052305330543055305630573058305930603061306230633064306530663067306830693070307130723073307430753076307730783079308030813082308330843085308630873088308930903091309230933094309530963097309830993100310131023103310431053106310731083109311031113112311331143115311631173118311931203121312231233124312531263127312831293130313131323133313431353136313731383139314031413142314331443145314631473148314931503151315231533154315531563157315831593160316131623163316431653166316731683169317031713172317331743175317631773178317931803181318231833184318531863187318831893190319131923193319431953196319731983199320032013202320332043205320632073208320932103211321232133214321532163217321832193220322132223223322432253226322732283229323032313232323332343235323632373238323932403241324232433244324532463247324832493250325132523253325432553256325732583259326032613262326332643265326632673268326932703271327232733274327532763277327832793280328132823283328432853286328732883289329032913292329332943295329632973298329933003301330233033304330533063307330833093310331133123313331433153316331733183319332033213322332333243325332633273328332933303331333233333334333533363337333833393340334133423343334433453346334733483349335033513352335333543355335633573358335933603361336233633364336533663367336833693370337133723373337433753376337733783379338033813382338333843385338633873388338933903391339233933394339533963397339833993400340134023403340434053406340734083409341034113412341334143415341634173418341934203421342234233424342534263427342834293430343134323433343434353436343734383439344034413442344334443445344634473448344934503451345234533454345534563457345834593460346134623463346434653466346734683469347034713472347334743475347634773478347934803481348234833484348534863487348834893490349134923493349434953496349734983499350035013502350335043505350635073508350935103511351235133514351535163517351835193520352135223523352435253526352735283529353035313532353335343535353635373538353935403541354235433544354535463547354835493550355135523553355435553556355735583559356035613562356335643565356635673568356935703571357235733574357535763577357835793580358135823583358435853586358735883589359035913592359335943595359635973598359936003601360236033604360536063607360836093610361136123613361436153616361736183619362036213622362336243625362636273628362936303631363236333634363536363637363836393640364136423643364436453646364736483649365036513652365336543655365636573658365936603661366236633664366536663667366836693670367136723673367436753676367736783679368036813682368336843685368636873688368936903691369236933694369536963697369836993700370137023703370437053706370737083709371037113712371337143715371637173718371937203721372237233724372537263727372837293730373137323733373437353736373737383739374037413742374337443745374637473748374937503751375237533754375537563757375837593760376137623763376437653766376737683769377037713772377337743775377637773778377937803781378237833784378537863787378837893790379137923793379437953796379737983799380038013802380338043805380638073808380938103811381238133814381538163817381838193820382138223823382438253826382738283829383038313832383338343835383638373838383938403841384238433844384538463847384838493850385138523853385438553856385738583859386038613862386338643865386638673868386938703871387238733874387538763877387838793880388138823883388438853886388738883889389038913892389338943895389638973898389939003901390239033904390539063907390839093910391139123913391439153916391739183919392039213922392339243925392639273928392939303931393239333934393539363937393839393940394139423943394439453946394739483949395039513952395339543955395639573958395939603961396239633964396539663967396839693970397139723973397439753976397739783979398039813982398339843985398639873988398939903991399239933994399539963997399839994000400140024003400440054006400740084009401040114012401340144015401640174018401940204021402240234024402540264027402840294030403140324033403440354036403740384039404040414042404340444045404640474048404940504051405240534054405540564057405840594060406140624063406440654066406740684069407040714072407340744075407640774078407940804081408240834084408540864087408840894090409140924093409440954096409740984099410041014102410341044105410641074108410941104111411241134114411541164117411841194120412141224123412441254126412741284129413041314132413341344135413641374138413941404141414241434144414541464147414841494150415141524153415441554156415741584159416041614162416341644165416641674168416941704171417241734174417541764177417841794180418141824183418441854186418741884189419041914192419341944195419641974198419942004201420242034204420542064207420842094210421142124213421442154216421742184219422042214222422342244225422642274228422942304231423242334234423542364237423842394240424142424243424442454246424742484249425042514252425342544255425642574258425942604261426242634264426542664267426842694270427142724273427442754276427742784279428042814282428342844285428642874288428942904291429242934294429542964297429842994300430143024303430443054306430743084309431043114312431343144315431643174318431943204321432243234324432543264327432843294330433143324333433443354336433743384339434043414342434343444345434643474348434943504351435243534354435543564357435843594360436143624363436443654366436743684369437043714372437343744375437643774378437943804381438243834384438543864387438843894390439143924393439443954396439743984399440044014402440344044405440644074408440944104411441244134414441544164417441844194420442144224423442444254426442744284429443044314432443344344435443644374438443944404441444244434444444544464447444844494450445144524453445444554456445744584459446044614462446344644465446644674468446944704471447244734474447544764477447844794480448144824483448444854486448744884489449044914492449344944495449644974498449945004501450245034504450545064507450845094510451145124513451445154516451745184519452045214522452345244525452645274528452945304531453245334534453545364537453845394540454145424543454445454546454745484549455045514552455345544555455645574558455945604561456245634564456545664567456845694570457145724573457445754576457745784579458045814582458345844585458645874588458945904591459245934594459545964597459845994600460146024603460446054606460746084609461046114612461346144615461646174618461946204621462246234624462546264627462846294630463146324633463446354636463746384639464046414642464346444645464646474648464946504651465246534654465546564657465846594660466146624663466446654666466746684669467046714672467346744675467646774678467946804681468246834684468546864687468846894690469146924693469446954696469746984699470047014702470347044705470647074708470947104711471247134714471547164717471847194720472147224723472447254726472747284729473047314732473347344735473647374738473947404741474247434744474547464747474847494750475147524753475447554756475747584759476047614762476347644765476647674768476947704771477247734774477547764777477847794780478147824783478447854786478747884789479047914792479347944795479647974798479948004801480248034804480548064807480848094810481148124813481448154816481748184819482048214822482348244825482648274828482948304831483248334834483548364837483848394840484148424843484448454846484748484849485048514852485348544855485648574858485948604861486248634864486548664867486848694870487148724873487448754876487748784879488048814882488348844885488648874888488948904891489248934894489548964897489848994900490149024903490449054906490749084909491049114912491349144915491649174918491949204921492249234924492549264927492849294930493149324933493449354936493749384939494049414942494349444945494649474948494949504951495249534954495549564957495849594960496149624963496449654966496749684969497049714972497349744975497649774978497949804981498249834984498549864987498849894990499149924993499449954996499749984999500050015002500350045005500650075008500950105011501250135014501550165017501850195020502150225023502450255026502750285029503050315032503350345035503650375038503950405041504250435044504550465047504850495050505150525053505450555056505750585059506050615062506350645065506650675068506950705071507250735074507550765077507850795080508150825083508450855086508750885089509050915092509350945095509650975098509951005101510251035104510551065107510851095110511151125113511451155116511751185119512051215122512351245125512651275128512951305131513251335134513551365137513851395140514151425143514451455146514751485149515051515152515351545155515651575158515951605161516251635164516551665167516851695170517151725173517451755176517751785179518051815182518351845185518651875188518951905191519251935194519551965197519851995200520152025203520452055206520752085209521052115212521352145215521652175218521952205221522252235224
  1. // SPDX-License-Identifier: GPL-2.0
  2. #include <linux/ceph/ceph_debug.h>
  3. #include <linux/fs.h>
  4. #include <linux/wait.h>
  5. #include <linux/slab.h>
  6. #include <linux/gfp.h>
  7. #include <linux/sched.h>
  8. #include <linux/debugfs.h>
  9. #include <linux/seq_file.h>
  10. #include <linux/ratelimit.h>
  11. #include <linux/bits.h>
  12. #include <linux/ktime.h>
  13. #include "super.h"
  14. #include "mds_client.h"
  15. #include <linux/ceph/ceph_features.h>
  16. #include <linux/ceph/messenger.h>
  17. #include <linux/ceph/decode.h>
  18. #include <linux/ceph/pagelist.h>
  19. #include <linux/ceph/auth.h>
  20. #include <linux/ceph/debugfs.h>
  21. #define RECONNECT_MAX_SIZE (INT_MAX - PAGE_SIZE)
  22. /*
  23. * A cluster of MDS (metadata server) daemons is responsible for
  24. * managing the file system namespace (the directory hierarchy and
  25. * inodes) and for coordinating shared access to storage. Metadata is
  26. * partitioning hierarchically across a number of servers, and that
  27. * partition varies over time as the cluster adjusts the distribution
  28. * in order to balance load.
  29. *
  30. * The MDS client is primarily responsible to managing synchronous
  31. * metadata requests for operations like open, unlink, and so forth.
  32. * If there is a MDS failure, we find out about it when we (possibly
  33. * request and) receive a new MDS map, and can resubmit affected
  34. * requests.
  35. *
  36. * For the most part, though, we take advantage of a lossless
  37. * communications channel to the MDS, and do not need to worry about
  38. * timing out or resubmitting requests.
  39. *
  40. * We maintain a stateful "session" with each MDS we interact with.
  41. * Within each session, we sent periodic heartbeat messages to ensure
  42. * any capabilities or leases we have been issues remain valid. If
  43. * the session times out and goes stale, our leases and capabilities
  44. * are no longer valid.
  45. */
  46. struct ceph_reconnect_state {
  47. struct ceph_mds_session *session;
  48. int nr_caps, nr_realms;
  49. struct ceph_pagelist *pagelist;
  50. unsigned msg_version;
  51. bool allow_multi;
  52. };
  53. static void __wake_requests(struct ceph_mds_client *mdsc,
  54. struct list_head *head);
  55. static void ceph_cap_release_work(struct work_struct *work);
  56. static void ceph_cap_reclaim_work(struct work_struct *work);
  57. static const struct ceph_connection_operations mds_con_ops;
  58. /*
  59. * mds reply parsing
  60. */
  61. static int parse_reply_info_quota(void **p, void *end,
  62. struct ceph_mds_reply_info_in *info)
  63. {
  64. u8 struct_v, struct_compat;
  65. u32 struct_len;
  66. ceph_decode_8_safe(p, end, struct_v, bad);
  67. ceph_decode_8_safe(p, end, struct_compat, bad);
  68. /* struct_v is expected to be >= 1. we only
  69. * understand encoding with struct_compat == 1. */
  70. if (!struct_v || struct_compat != 1)
  71. goto bad;
  72. ceph_decode_32_safe(p, end, struct_len, bad);
  73. ceph_decode_need(p, end, struct_len, bad);
  74. end = *p + struct_len;
  75. ceph_decode_64_safe(p, end, info->max_bytes, bad);
  76. ceph_decode_64_safe(p, end, info->max_files, bad);
  77. *p = end;
  78. return 0;
  79. bad:
  80. return -EIO;
  81. }
  82. /*
  83. * parse individual inode info
  84. */
  85. static int parse_reply_info_in(void **p, void *end,
  86. struct ceph_mds_reply_info_in *info,
  87. u64 features)
  88. {
  89. int err = 0;
  90. u8 struct_v = 0;
  91. if (features == (u64)-1) {
  92. u32 struct_len;
  93. u8 struct_compat;
  94. ceph_decode_8_safe(p, end, struct_v, bad);
  95. ceph_decode_8_safe(p, end, struct_compat, bad);
  96. /* struct_v is expected to be >= 1. we only understand
  97. * encoding with struct_compat == 1. */
  98. if (!struct_v || struct_compat != 1)
  99. goto bad;
  100. ceph_decode_32_safe(p, end, struct_len, bad);
  101. ceph_decode_need(p, end, struct_len, bad);
  102. end = *p + struct_len;
  103. }
  104. ceph_decode_need(p, end, sizeof(struct ceph_mds_reply_inode), bad);
  105. info->in = *p;
  106. *p += sizeof(struct ceph_mds_reply_inode) +
  107. sizeof(*info->in->fragtree.splits) *
  108. le32_to_cpu(info->in->fragtree.nsplits);
  109. ceph_decode_32_safe(p, end, info->symlink_len, bad);
  110. ceph_decode_need(p, end, info->symlink_len, bad);
  111. info->symlink = *p;
  112. *p += info->symlink_len;
  113. ceph_decode_copy_safe(p, end, &info->dir_layout,
  114. sizeof(info->dir_layout), bad);
  115. ceph_decode_32_safe(p, end, info->xattr_len, bad);
  116. ceph_decode_need(p, end, info->xattr_len, bad);
  117. info->xattr_data = *p;
  118. *p += info->xattr_len;
  119. if (features == (u64)-1) {
  120. /* inline data */
  121. ceph_decode_64_safe(p, end, info->inline_version, bad);
  122. ceph_decode_32_safe(p, end, info->inline_len, bad);
  123. ceph_decode_need(p, end, info->inline_len, bad);
  124. info->inline_data = *p;
  125. *p += info->inline_len;
  126. /* quota */
  127. err = parse_reply_info_quota(p, end, info);
  128. if (err < 0)
  129. goto out_bad;
  130. /* pool namespace */
  131. ceph_decode_32_safe(p, end, info->pool_ns_len, bad);
  132. if (info->pool_ns_len > 0) {
  133. ceph_decode_need(p, end, info->pool_ns_len, bad);
  134. info->pool_ns_data = *p;
  135. *p += info->pool_ns_len;
  136. }
  137. /* btime */
  138. ceph_decode_need(p, end, sizeof(info->btime), bad);
  139. ceph_decode_copy(p, &info->btime, sizeof(info->btime));
  140. /* change attribute */
  141. ceph_decode_64_safe(p, end, info->change_attr, bad);
  142. /* dir pin */
  143. if (struct_v >= 2) {
  144. ceph_decode_32_safe(p, end, info->dir_pin, bad);
  145. } else {
  146. info->dir_pin = -ENODATA;
  147. }
  148. /* snapshot birth time, remains zero for v<=2 */
  149. if (struct_v >= 3) {
  150. ceph_decode_need(p, end, sizeof(info->snap_btime), bad);
  151. ceph_decode_copy(p, &info->snap_btime,
  152. sizeof(info->snap_btime));
  153. } else {
  154. memset(&info->snap_btime, 0, sizeof(info->snap_btime));
  155. }
  156. *p = end;
  157. } else {
  158. if (features & CEPH_FEATURE_MDS_INLINE_DATA) {
  159. ceph_decode_64_safe(p, end, info->inline_version, bad);
  160. ceph_decode_32_safe(p, end, info->inline_len, bad);
  161. ceph_decode_need(p, end, info->inline_len, bad);
  162. info->inline_data = *p;
  163. *p += info->inline_len;
  164. } else
  165. info->inline_version = CEPH_INLINE_NONE;
  166. if (features & CEPH_FEATURE_MDS_QUOTA) {
  167. err = parse_reply_info_quota(p, end, info);
  168. if (err < 0)
  169. goto out_bad;
  170. } else {
  171. info->max_bytes = 0;
  172. info->max_files = 0;
  173. }
  174. info->pool_ns_len = 0;
  175. info->pool_ns_data = NULL;
  176. if (features & CEPH_FEATURE_FS_FILE_LAYOUT_V2) {
  177. ceph_decode_32_safe(p, end, info->pool_ns_len, bad);
  178. if (info->pool_ns_len > 0) {
  179. ceph_decode_need(p, end, info->pool_ns_len, bad);
  180. info->pool_ns_data = *p;
  181. *p += info->pool_ns_len;
  182. }
  183. }
  184. if (features & CEPH_FEATURE_FS_BTIME) {
  185. ceph_decode_need(p, end, sizeof(info->btime), bad);
  186. ceph_decode_copy(p, &info->btime, sizeof(info->btime));
  187. ceph_decode_64_safe(p, end, info->change_attr, bad);
  188. }
  189. info->dir_pin = -ENODATA;
  190. /* info->snap_btime remains zero */
  191. }
  192. return 0;
  193. bad:
  194. err = -EIO;
  195. out_bad:
  196. return err;
  197. }
  198. static int parse_reply_info_dir(void **p, void *end,
  199. struct ceph_mds_reply_dirfrag **dirfrag,
  200. u64 features)
  201. {
  202. if (features == (u64)-1) {
  203. u8 struct_v, struct_compat;
  204. u32 struct_len;
  205. ceph_decode_8_safe(p, end, struct_v, bad);
  206. ceph_decode_8_safe(p, end, struct_compat, bad);
  207. /* struct_v is expected to be >= 1. we only understand
  208. * encoding whose struct_compat == 1. */
  209. if (!struct_v || struct_compat != 1)
  210. goto bad;
  211. ceph_decode_32_safe(p, end, struct_len, bad);
  212. ceph_decode_need(p, end, struct_len, bad);
  213. end = *p + struct_len;
  214. }
  215. ceph_decode_need(p, end, sizeof(**dirfrag), bad);
  216. *dirfrag = *p;
  217. *p += sizeof(**dirfrag) + sizeof(u32) * le32_to_cpu((*dirfrag)->ndist);
  218. if (unlikely(*p > end))
  219. goto bad;
  220. if (features == (u64)-1)
  221. *p = end;
  222. return 0;
  223. bad:
  224. return -EIO;
  225. }
  226. static int parse_reply_info_lease(void **p, void *end,
  227. struct ceph_mds_reply_lease **lease,
  228. u64 features)
  229. {
  230. if (features == (u64)-1) {
  231. u8 struct_v, struct_compat;
  232. u32 struct_len;
  233. ceph_decode_8_safe(p, end, struct_v, bad);
  234. ceph_decode_8_safe(p, end, struct_compat, bad);
  235. /* struct_v is expected to be >= 1. we only understand
  236. * encoding whose struct_compat == 1. */
  237. if (!struct_v || struct_compat != 1)
  238. goto bad;
  239. ceph_decode_32_safe(p, end, struct_len, bad);
  240. ceph_decode_need(p, end, struct_len, bad);
  241. end = *p + struct_len;
  242. }
  243. ceph_decode_need(p, end, sizeof(**lease), bad);
  244. *lease = *p;
  245. *p += sizeof(**lease);
  246. if (features == (u64)-1)
  247. *p = end;
  248. return 0;
  249. bad:
  250. return -EIO;
  251. }
  252. /*
  253. * parse a normal reply, which may contain a (dir+)dentry and/or a
  254. * target inode.
  255. */
  256. static int parse_reply_info_trace(void **p, void *end,
  257. struct ceph_mds_reply_info_parsed *info,
  258. u64 features)
  259. {
  260. int err;
  261. if (info->head->is_dentry) {
  262. err = parse_reply_info_in(p, end, &info->diri, features);
  263. if (err < 0)
  264. goto out_bad;
  265. err = parse_reply_info_dir(p, end, &info->dirfrag, features);
  266. if (err < 0)
  267. goto out_bad;
  268. ceph_decode_32_safe(p, end, info->dname_len, bad);
  269. ceph_decode_need(p, end, info->dname_len, bad);
  270. info->dname = *p;
  271. *p += info->dname_len;
  272. err = parse_reply_info_lease(p, end, &info->dlease, features);
  273. if (err < 0)
  274. goto out_bad;
  275. }
  276. if (info->head->is_target) {
  277. err = parse_reply_info_in(p, end, &info->targeti, features);
  278. if (err < 0)
  279. goto out_bad;
  280. }
  281. if (unlikely(*p != end))
  282. goto bad;
  283. return 0;
  284. bad:
  285. err = -EIO;
  286. out_bad:
  287. pr_err("problem parsing mds trace %d\n", err);
  288. return err;
  289. }
  290. /*
  291. * parse readdir results
  292. */
  293. static int parse_reply_info_readdir(void **p, void *end,
  294. struct ceph_mds_reply_info_parsed *info,
  295. u64 features)
  296. {
  297. u32 num, i = 0;
  298. int err;
  299. err = parse_reply_info_dir(p, end, &info->dir_dir, features);
  300. if (err < 0)
  301. goto out_bad;
  302. ceph_decode_need(p, end, sizeof(num) + 2, bad);
  303. num = ceph_decode_32(p);
  304. {
  305. u16 flags = ceph_decode_16(p);
  306. info->dir_end = !!(flags & CEPH_READDIR_FRAG_END);
  307. info->dir_complete = !!(flags & CEPH_READDIR_FRAG_COMPLETE);
  308. info->hash_order = !!(flags & CEPH_READDIR_HASH_ORDER);
  309. info->offset_hash = !!(flags & CEPH_READDIR_OFFSET_HASH);
  310. }
  311. if (num == 0)
  312. goto done;
  313. BUG_ON(!info->dir_entries);
  314. if ((unsigned long)(info->dir_entries + num) >
  315. (unsigned long)info->dir_entries + info->dir_buf_size) {
  316. pr_err("dir contents are larger than expected\n");
  317. WARN_ON(1);
  318. goto bad;
  319. }
  320. info->dir_nr = num;
  321. while (num) {
  322. struct ceph_mds_reply_dir_entry *rde = info->dir_entries + i;
  323. /* dentry */
  324. ceph_decode_32_safe(p, end, rde->name_len, bad);
  325. ceph_decode_need(p, end, rde->name_len, bad);
  326. rde->name = *p;
  327. *p += rde->name_len;
  328. dout("parsed dir dname '%.*s'\n", rde->name_len, rde->name);
  329. /* dentry lease */
  330. err = parse_reply_info_lease(p, end, &rde->lease, features);
  331. if (err)
  332. goto out_bad;
  333. /* inode */
  334. err = parse_reply_info_in(p, end, &rde->inode, features);
  335. if (err < 0)
  336. goto out_bad;
  337. /* ceph_readdir_prepopulate() will update it */
  338. rde->offset = 0;
  339. i++;
  340. num--;
  341. }
  342. done:
  343. /* Skip over any unrecognized fields */
  344. *p = end;
  345. return 0;
  346. bad:
  347. err = -EIO;
  348. out_bad:
  349. pr_err("problem parsing dir contents %d\n", err);
  350. return err;
  351. }
  352. /*
  353. * parse fcntl F_GETLK results
  354. */
  355. static int parse_reply_info_filelock(void **p, void *end,
  356. struct ceph_mds_reply_info_parsed *info,
  357. u64 features)
  358. {
  359. if (*p + sizeof(*info->filelock_reply) > end)
  360. goto bad;
  361. info->filelock_reply = *p;
  362. /* Skip over any unrecognized fields */
  363. *p = end;
  364. return 0;
  365. bad:
  366. return -EIO;
  367. }
  368. #if BITS_PER_LONG == 64
  369. #define DELEGATED_INO_AVAILABLE xa_mk_value(1)
  370. static int ceph_parse_deleg_inos(void **p, void *end,
  371. struct ceph_mds_session *s)
  372. {
  373. u32 sets;
  374. ceph_decode_32_safe(p, end, sets, bad);
  375. dout("got %u sets of delegated inodes\n", sets);
  376. while (sets--) {
  377. u64 start, len, ino;
  378. ceph_decode_64_safe(p, end, start, bad);
  379. ceph_decode_64_safe(p, end, len, bad);
  380. /* Don't accept a delegation of system inodes */
  381. if (start < CEPH_INO_SYSTEM_BASE) {
  382. pr_warn_ratelimited("ceph: ignoring reserved inode range delegation (start=0x%llx len=0x%llx)\n",
  383. start, len);
  384. continue;
  385. }
  386. while (len--) {
  387. int err = xa_insert(&s->s_delegated_inos, ino = start++,
  388. DELEGATED_INO_AVAILABLE,
  389. GFP_KERNEL);
  390. if (!err) {
  391. dout("added delegated inode 0x%llx\n",
  392. start - 1);
  393. } else if (err == -EBUSY) {
  394. pr_warn("ceph: MDS delegated inode 0x%llx more than once.\n",
  395. start - 1);
  396. } else {
  397. return err;
  398. }
  399. }
  400. }
  401. return 0;
  402. bad:
  403. return -EIO;
  404. }
  405. u64 ceph_get_deleg_ino(struct ceph_mds_session *s)
  406. {
  407. unsigned long ino;
  408. void *val;
  409. xa_for_each(&s->s_delegated_inos, ino, val) {
  410. val = xa_erase(&s->s_delegated_inos, ino);
  411. if (val == DELEGATED_INO_AVAILABLE)
  412. return ino;
  413. }
  414. return 0;
  415. }
  416. int ceph_restore_deleg_ino(struct ceph_mds_session *s, u64 ino)
  417. {
  418. return xa_insert(&s->s_delegated_inos, ino, DELEGATED_INO_AVAILABLE,
  419. GFP_KERNEL);
  420. }
  421. #else /* BITS_PER_LONG == 64 */
  422. /*
  423. * FIXME: xarrays can't handle 64-bit indexes on a 32-bit arch. For now, just
  424. * ignore delegated_inos on 32 bit arch. Maybe eventually add xarrays for top
  425. * and bottom words?
  426. */
  427. static int ceph_parse_deleg_inos(void **p, void *end,
  428. struct ceph_mds_session *s)
  429. {
  430. u32 sets;
  431. ceph_decode_32_safe(p, end, sets, bad);
  432. if (sets)
  433. ceph_decode_skip_n(p, end, sets * 2 * sizeof(__le64), bad);
  434. return 0;
  435. bad:
  436. return -EIO;
  437. }
  438. u64 ceph_get_deleg_ino(struct ceph_mds_session *s)
  439. {
  440. return 0;
  441. }
  442. int ceph_restore_deleg_ino(struct ceph_mds_session *s, u64 ino)
  443. {
  444. return 0;
  445. }
  446. #endif /* BITS_PER_LONG == 64 */
  447. /*
  448. * parse create results
  449. */
  450. static int parse_reply_info_create(void **p, void *end,
  451. struct ceph_mds_reply_info_parsed *info,
  452. u64 features, struct ceph_mds_session *s)
  453. {
  454. int ret;
  455. if (features == (u64)-1 ||
  456. (features & CEPH_FEATURE_REPLY_CREATE_INODE)) {
  457. if (*p == end) {
  458. /* Malformed reply? */
  459. info->has_create_ino = false;
  460. } else if (test_bit(CEPHFS_FEATURE_DELEG_INO, &s->s_features)) {
  461. u8 struct_v, struct_compat;
  462. u32 len;
  463. info->has_create_ino = true;
  464. ceph_decode_8_safe(p, end, struct_v, bad);
  465. ceph_decode_8_safe(p, end, struct_compat, bad);
  466. ceph_decode_32_safe(p, end, len, bad);
  467. ceph_decode_64_safe(p, end, info->ino, bad);
  468. ret = ceph_parse_deleg_inos(p, end, s);
  469. if (ret)
  470. return ret;
  471. } else {
  472. /* legacy */
  473. ceph_decode_64_safe(p, end, info->ino, bad);
  474. info->has_create_ino = true;
  475. }
  476. } else {
  477. if (*p != end)
  478. goto bad;
  479. }
  480. /* Skip over any unrecognized fields */
  481. *p = end;
  482. return 0;
  483. bad:
  484. return -EIO;
  485. }
  486. /*
  487. * parse extra results
  488. */
  489. static int parse_reply_info_extra(void **p, void *end,
  490. struct ceph_mds_reply_info_parsed *info,
  491. u64 features, struct ceph_mds_session *s)
  492. {
  493. u32 op = le32_to_cpu(info->head->op);
  494. if (op == CEPH_MDS_OP_GETFILELOCK)
  495. return parse_reply_info_filelock(p, end, info, features);
  496. else if (op == CEPH_MDS_OP_READDIR || op == CEPH_MDS_OP_LSSNAP)
  497. return parse_reply_info_readdir(p, end, info, features);
  498. else if (op == CEPH_MDS_OP_CREATE)
  499. return parse_reply_info_create(p, end, info, features, s);
  500. else
  501. return -EIO;
  502. }
  503. /*
  504. * parse entire mds reply
  505. */
  506. static int parse_reply_info(struct ceph_mds_session *s, struct ceph_msg *msg,
  507. struct ceph_mds_reply_info_parsed *info,
  508. u64 features)
  509. {
  510. void *p, *end;
  511. u32 len;
  512. int err;
  513. info->head = msg->front.iov_base;
  514. p = msg->front.iov_base + sizeof(struct ceph_mds_reply_head);
  515. end = p + msg->front.iov_len - sizeof(struct ceph_mds_reply_head);
  516. /* trace */
  517. ceph_decode_32_safe(&p, end, len, bad);
  518. if (len > 0) {
  519. ceph_decode_need(&p, end, len, bad);
  520. err = parse_reply_info_trace(&p, p+len, info, features);
  521. if (err < 0)
  522. goto out_bad;
  523. }
  524. /* extra */
  525. ceph_decode_32_safe(&p, end, len, bad);
  526. if (len > 0) {
  527. ceph_decode_need(&p, end, len, bad);
  528. err = parse_reply_info_extra(&p, p+len, info, features, s);
  529. if (err < 0)
  530. goto out_bad;
  531. }
  532. /* snap blob */
  533. ceph_decode_32_safe(&p, end, len, bad);
  534. info->snapblob_len = len;
  535. info->snapblob = p;
  536. p += len;
  537. if (p != end)
  538. goto bad;
  539. return 0;
  540. bad:
  541. err = -EIO;
  542. out_bad:
  543. pr_err("mds parse_reply err %d\n", err);
  544. return err;
  545. }
  546. static void destroy_reply_info(struct ceph_mds_reply_info_parsed *info)
  547. {
  548. if (!info->dir_entries)
  549. return;
  550. free_pages((unsigned long)info->dir_entries, get_order(info->dir_buf_size));
  551. }
  552. /*
  553. * sessions
  554. */
  555. const char *ceph_session_state_name(int s)
  556. {
  557. switch (s) {
  558. case CEPH_MDS_SESSION_NEW: return "new";
  559. case CEPH_MDS_SESSION_OPENING: return "opening";
  560. case CEPH_MDS_SESSION_OPEN: return "open";
  561. case CEPH_MDS_SESSION_HUNG: return "hung";
  562. case CEPH_MDS_SESSION_CLOSING: return "closing";
  563. case CEPH_MDS_SESSION_CLOSED: return "closed";
  564. case CEPH_MDS_SESSION_RESTARTING: return "restarting";
  565. case CEPH_MDS_SESSION_RECONNECTING: return "reconnecting";
  566. case CEPH_MDS_SESSION_REJECTED: return "rejected";
  567. default: return "???";
  568. }
  569. }
  570. struct ceph_mds_session *ceph_get_mds_session(struct ceph_mds_session *s)
  571. {
  572. if (refcount_inc_not_zero(&s->s_ref)) {
  573. dout("mdsc get_session %p %d -> %d\n", s,
  574. refcount_read(&s->s_ref)-1, refcount_read(&s->s_ref));
  575. return s;
  576. } else {
  577. dout("mdsc get_session %p 0 -- FAIL\n", s);
  578. return NULL;
  579. }
  580. }
  581. void ceph_put_mds_session(struct ceph_mds_session *s)
  582. {
  583. if (IS_ERR_OR_NULL(s))
  584. return;
  585. dout("mdsc put_session %p %d -> %d\n", s,
  586. refcount_read(&s->s_ref), refcount_read(&s->s_ref)-1);
  587. if (refcount_dec_and_test(&s->s_ref)) {
  588. if (s->s_auth.authorizer)
  589. ceph_auth_destroy_authorizer(s->s_auth.authorizer);
  590. WARN_ON(mutex_is_locked(&s->s_mutex));
  591. xa_destroy(&s->s_delegated_inos);
  592. kfree(s);
  593. }
  594. }
  595. /*
  596. * called under mdsc->mutex
  597. */
  598. struct ceph_mds_session *__ceph_lookup_mds_session(struct ceph_mds_client *mdsc,
  599. int mds)
  600. {
  601. if (mds >= mdsc->max_sessions || !mdsc->sessions[mds])
  602. return NULL;
  603. return ceph_get_mds_session(mdsc->sessions[mds]);
  604. }
  605. static bool __have_session(struct ceph_mds_client *mdsc, int mds)
  606. {
  607. if (mds >= mdsc->max_sessions || !mdsc->sessions[mds])
  608. return false;
  609. else
  610. return true;
  611. }
  612. static int __verify_registered_session(struct ceph_mds_client *mdsc,
  613. struct ceph_mds_session *s)
  614. {
  615. if (s->s_mds >= mdsc->max_sessions ||
  616. mdsc->sessions[s->s_mds] != s)
  617. return -ENOENT;
  618. return 0;
  619. }
  620. /*
  621. * create+register a new session for given mds.
  622. * called under mdsc->mutex.
  623. */
  624. static struct ceph_mds_session *register_session(struct ceph_mds_client *mdsc,
  625. int mds)
  626. {
  627. struct ceph_mds_session *s;
  628. if (mds >= mdsc->mdsmap->possible_max_rank)
  629. return ERR_PTR(-EINVAL);
  630. s = kzalloc(sizeof(*s), GFP_NOFS);
  631. if (!s)
  632. return ERR_PTR(-ENOMEM);
  633. if (mds >= mdsc->max_sessions) {
  634. int newmax = 1 << get_count_order(mds + 1);
  635. struct ceph_mds_session **sa;
  636. dout("%s: realloc to %d\n", __func__, newmax);
  637. sa = kcalloc(newmax, sizeof(void *), GFP_NOFS);
  638. if (!sa)
  639. goto fail_realloc;
  640. if (mdsc->sessions) {
  641. memcpy(sa, mdsc->sessions,
  642. mdsc->max_sessions * sizeof(void *));
  643. kfree(mdsc->sessions);
  644. }
  645. mdsc->sessions = sa;
  646. mdsc->max_sessions = newmax;
  647. }
  648. dout("%s: mds%d\n", __func__, mds);
  649. s->s_mdsc = mdsc;
  650. s->s_mds = mds;
  651. s->s_state = CEPH_MDS_SESSION_NEW;
  652. s->s_ttl = 0;
  653. s->s_seq = 0;
  654. mutex_init(&s->s_mutex);
  655. ceph_con_init(&s->s_con, s, &mds_con_ops, &mdsc->fsc->client->msgr);
  656. spin_lock_init(&s->s_gen_ttl_lock);
  657. s->s_cap_gen = 1;
  658. s->s_cap_ttl = jiffies - 1;
  659. spin_lock_init(&s->s_cap_lock);
  660. s->s_renew_requested = 0;
  661. s->s_renew_seq = 0;
  662. INIT_LIST_HEAD(&s->s_caps);
  663. s->s_nr_caps = 0;
  664. refcount_set(&s->s_ref, 1);
  665. INIT_LIST_HEAD(&s->s_waiting);
  666. INIT_LIST_HEAD(&s->s_unsafe);
  667. xa_init(&s->s_delegated_inos);
  668. s->s_num_cap_releases = 0;
  669. s->s_cap_reconnect = 0;
  670. s->s_cap_iterator = NULL;
  671. INIT_LIST_HEAD(&s->s_cap_releases);
  672. INIT_WORK(&s->s_cap_release_work, ceph_cap_release_work);
  673. INIT_LIST_HEAD(&s->s_cap_dirty);
  674. INIT_LIST_HEAD(&s->s_cap_flushing);
  675. mdsc->sessions[mds] = s;
  676. atomic_inc(&mdsc->num_sessions);
  677. refcount_inc(&s->s_ref); /* one ref to sessions[], one to caller */
  678. ceph_con_open(&s->s_con, CEPH_ENTITY_TYPE_MDS, mds,
  679. ceph_mdsmap_get_addr(mdsc->mdsmap, mds));
  680. return s;
  681. fail_realloc:
  682. kfree(s);
  683. return ERR_PTR(-ENOMEM);
  684. }
  685. /*
  686. * called under mdsc->mutex
  687. */
  688. static void __unregister_session(struct ceph_mds_client *mdsc,
  689. struct ceph_mds_session *s)
  690. {
  691. dout("__unregister_session mds%d %p\n", s->s_mds, s);
  692. BUG_ON(mdsc->sessions[s->s_mds] != s);
  693. mdsc->sessions[s->s_mds] = NULL;
  694. ceph_con_close(&s->s_con);
  695. ceph_put_mds_session(s);
  696. atomic_dec(&mdsc->num_sessions);
  697. }
  698. /*
  699. * drop session refs in request.
  700. *
  701. * should be last request ref, or hold mdsc->mutex
  702. */
  703. static void put_request_session(struct ceph_mds_request *req)
  704. {
  705. if (req->r_session) {
  706. ceph_put_mds_session(req->r_session);
  707. req->r_session = NULL;
  708. }
  709. }
  710. void ceph_mdsc_release_request(struct kref *kref)
  711. {
  712. struct ceph_mds_request *req = container_of(kref,
  713. struct ceph_mds_request,
  714. r_kref);
  715. ceph_mdsc_release_dir_caps_no_check(req);
  716. destroy_reply_info(&req->r_reply_info);
  717. if (req->r_request)
  718. ceph_msg_put(req->r_request);
  719. if (req->r_reply)
  720. ceph_msg_put(req->r_reply);
  721. if (req->r_inode) {
  722. ceph_put_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
  723. /* avoid calling iput_final() in mds dispatch threads */
  724. ceph_async_iput(req->r_inode);
  725. }
  726. if (req->r_parent) {
  727. ceph_put_cap_refs(ceph_inode(req->r_parent), CEPH_CAP_PIN);
  728. ceph_async_iput(req->r_parent);
  729. }
  730. ceph_async_iput(req->r_target_inode);
  731. if (req->r_dentry)
  732. dput(req->r_dentry);
  733. if (req->r_old_dentry)
  734. dput(req->r_old_dentry);
  735. if (req->r_old_dentry_dir) {
  736. /*
  737. * track (and drop pins for) r_old_dentry_dir
  738. * separately, since r_old_dentry's d_parent may have
  739. * changed between the dir mutex being dropped and
  740. * this request being freed.
  741. */
  742. ceph_put_cap_refs(ceph_inode(req->r_old_dentry_dir),
  743. CEPH_CAP_PIN);
  744. ceph_async_iput(req->r_old_dentry_dir);
  745. }
  746. kfree(req->r_path1);
  747. kfree(req->r_path2);
  748. if (req->r_pagelist)
  749. ceph_pagelist_release(req->r_pagelist);
  750. put_request_session(req);
  751. ceph_unreserve_caps(req->r_mdsc, &req->r_caps_reservation);
  752. WARN_ON_ONCE(!list_empty(&req->r_wait));
  753. kmem_cache_free(ceph_mds_request_cachep, req);
  754. }
  755. DEFINE_RB_FUNCS(request, struct ceph_mds_request, r_tid, r_node)
  756. /*
  757. * lookup session, bump ref if found.
  758. *
  759. * called under mdsc->mutex.
  760. */
  761. static struct ceph_mds_request *
  762. lookup_get_request(struct ceph_mds_client *mdsc, u64 tid)
  763. {
  764. struct ceph_mds_request *req;
  765. req = lookup_request(&mdsc->request_tree, tid);
  766. if (req)
  767. ceph_mdsc_get_request(req);
  768. return req;
  769. }
  770. /*
  771. * Register an in-flight request, and assign a tid. Link to directory
  772. * are modifying (if any).
  773. *
  774. * Called under mdsc->mutex.
  775. */
  776. static void __register_request(struct ceph_mds_client *mdsc,
  777. struct ceph_mds_request *req,
  778. struct inode *dir)
  779. {
  780. int ret = 0;
  781. req->r_tid = ++mdsc->last_tid;
  782. if (req->r_num_caps) {
  783. ret = ceph_reserve_caps(mdsc, &req->r_caps_reservation,
  784. req->r_num_caps);
  785. if (ret < 0) {
  786. pr_err("__register_request %p "
  787. "failed to reserve caps: %d\n", req, ret);
  788. /* set req->r_err to fail early from __do_request */
  789. req->r_err = ret;
  790. return;
  791. }
  792. }
  793. dout("__register_request %p tid %lld\n", req, req->r_tid);
  794. ceph_mdsc_get_request(req);
  795. insert_request(&mdsc->request_tree, req);
  796. req->r_uid = current_fsuid();
  797. req->r_gid = current_fsgid();
  798. if (mdsc->oldest_tid == 0 && req->r_op != CEPH_MDS_OP_SETFILELOCK)
  799. mdsc->oldest_tid = req->r_tid;
  800. if (dir) {
  801. struct ceph_inode_info *ci = ceph_inode(dir);
  802. ihold(dir);
  803. req->r_unsafe_dir = dir;
  804. spin_lock(&ci->i_unsafe_lock);
  805. list_add_tail(&req->r_unsafe_dir_item, &ci->i_unsafe_dirops);
  806. spin_unlock(&ci->i_unsafe_lock);
  807. }
  808. }
  809. static void __unregister_request(struct ceph_mds_client *mdsc,
  810. struct ceph_mds_request *req)
  811. {
  812. dout("__unregister_request %p tid %lld\n", req, req->r_tid);
  813. /* Never leave an unregistered request on an unsafe list! */
  814. list_del_init(&req->r_unsafe_item);
  815. if (req->r_tid == mdsc->oldest_tid) {
  816. struct rb_node *p = rb_next(&req->r_node);
  817. mdsc->oldest_tid = 0;
  818. while (p) {
  819. struct ceph_mds_request *next_req =
  820. rb_entry(p, struct ceph_mds_request, r_node);
  821. if (next_req->r_op != CEPH_MDS_OP_SETFILELOCK) {
  822. mdsc->oldest_tid = next_req->r_tid;
  823. break;
  824. }
  825. p = rb_next(p);
  826. }
  827. }
  828. erase_request(&mdsc->request_tree, req);
  829. if (req->r_unsafe_dir) {
  830. struct ceph_inode_info *ci = ceph_inode(req->r_unsafe_dir);
  831. spin_lock(&ci->i_unsafe_lock);
  832. list_del_init(&req->r_unsafe_dir_item);
  833. spin_unlock(&ci->i_unsafe_lock);
  834. }
  835. if (req->r_target_inode &&
  836. test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
  837. struct ceph_inode_info *ci = ceph_inode(req->r_target_inode);
  838. spin_lock(&ci->i_unsafe_lock);
  839. list_del_init(&req->r_unsafe_target_item);
  840. spin_unlock(&ci->i_unsafe_lock);
  841. }
  842. if (req->r_unsafe_dir) {
  843. /* avoid calling iput_final() in mds dispatch threads */
  844. ceph_async_iput(req->r_unsafe_dir);
  845. req->r_unsafe_dir = NULL;
  846. }
  847. complete_all(&req->r_safe_completion);
  848. ceph_mdsc_put_request(req);
  849. }
  850. /*
  851. * Walk back up the dentry tree until we hit a dentry representing a
  852. * non-snapshot inode. We do this using the rcu_read_lock (which must be held
  853. * when calling this) to ensure that the objects won't disappear while we're
  854. * working with them. Once we hit a candidate dentry, we attempt to take a
  855. * reference to it, and return that as the result.
  856. */
  857. static struct inode *get_nonsnap_parent(struct dentry *dentry)
  858. {
  859. struct inode *inode = NULL;
  860. while (dentry && !IS_ROOT(dentry)) {
  861. inode = d_inode_rcu(dentry);
  862. if (!inode || ceph_snap(inode) == CEPH_NOSNAP)
  863. break;
  864. dentry = dentry->d_parent;
  865. }
  866. if (inode)
  867. inode = igrab(inode);
  868. return inode;
  869. }
  870. /*
  871. * Choose mds to send request to next. If there is a hint set in the
  872. * request (e.g., due to a prior forward hint from the mds), use that.
  873. * Otherwise, consult frag tree and/or caps to identify the
  874. * appropriate mds. If all else fails, choose randomly.
  875. *
  876. * Called under mdsc->mutex.
  877. */
  878. static int __choose_mds(struct ceph_mds_client *mdsc,
  879. struct ceph_mds_request *req,
  880. bool *random)
  881. {
  882. struct inode *inode;
  883. struct ceph_inode_info *ci;
  884. struct ceph_cap *cap;
  885. int mode = req->r_direct_mode;
  886. int mds = -1;
  887. u32 hash = req->r_direct_hash;
  888. bool is_hash = test_bit(CEPH_MDS_R_DIRECT_IS_HASH, &req->r_req_flags);
  889. if (random)
  890. *random = false;
  891. /*
  892. * is there a specific mds we should try? ignore hint if we have
  893. * no session and the mds is not up (active or recovering).
  894. */
  895. if (req->r_resend_mds >= 0 &&
  896. (__have_session(mdsc, req->r_resend_mds) ||
  897. ceph_mdsmap_get_state(mdsc->mdsmap, req->r_resend_mds) > 0)) {
  898. dout("%s using resend_mds mds%d\n", __func__,
  899. req->r_resend_mds);
  900. return req->r_resend_mds;
  901. }
  902. if (mode == USE_RANDOM_MDS)
  903. goto random;
  904. inode = NULL;
  905. if (req->r_inode) {
  906. if (ceph_snap(req->r_inode) != CEPH_SNAPDIR) {
  907. inode = req->r_inode;
  908. ihold(inode);
  909. } else {
  910. /* req->r_dentry is non-null for LSSNAP request */
  911. rcu_read_lock();
  912. inode = get_nonsnap_parent(req->r_dentry);
  913. rcu_read_unlock();
  914. dout("%s using snapdir's parent %p\n", __func__, inode);
  915. }
  916. } else if (req->r_dentry) {
  917. /* ignore race with rename; old or new d_parent is okay */
  918. struct dentry *parent;
  919. struct inode *dir;
  920. rcu_read_lock();
  921. parent = READ_ONCE(req->r_dentry->d_parent);
  922. dir = req->r_parent ? : d_inode_rcu(parent);
  923. if (!dir || dir->i_sb != mdsc->fsc->sb) {
  924. /* not this fs or parent went negative */
  925. inode = d_inode(req->r_dentry);
  926. if (inode)
  927. ihold(inode);
  928. } else if (ceph_snap(dir) != CEPH_NOSNAP) {
  929. /* direct snapped/virtual snapdir requests
  930. * based on parent dir inode */
  931. inode = get_nonsnap_parent(parent);
  932. dout("%s using nonsnap parent %p\n", __func__, inode);
  933. } else {
  934. /* dentry target */
  935. inode = d_inode(req->r_dentry);
  936. if (!inode || mode == USE_AUTH_MDS) {
  937. /* dir + name */
  938. inode = igrab(dir);
  939. hash = ceph_dentry_hash(dir, req->r_dentry);
  940. is_hash = true;
  941. } else {
  942. ihold(inode);
  943. }
  944. }
  945. rcu_read_unlock();
  946. }
  947. dout("%s %p is_hash=%d (0x%x) mode %d\n", __func__, inode, (int)is_hash,
  948. hash, mode);
  949. if (!inode)
  950. goto random;
  951. ci = ceph_inode(inode);
  952. if (is_hash && S_ISDIR(inode->i_mode)) {
  953. struct ceph_inode_frag frag;
  954. int found;
  955. ceph_choose_frag(ci, hash, &frag, &found);
  956. if (found) {
  957. if (mode == USE_ANY_MDS && frag.ndist > 0) {
  958. u8 r;
  959. /* choose a random replica */
  960. get_random_bytes(&r, 1);
  961. r %= frag.ndist;
  962. mds = frag.dist[r];
  963. dout("%s %p %llx.%llx frag %u mds%d (%d/%d)\n",
  964. __func__, inode, ceph_vinop(inode),
  965. frag.frag, mds, (int)r, frag.ndist);
  966. if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=
  967. CEPH_MDS_STATE_ACTIVE &&
  968. !ceph_mdsmap_is_laggy(mdsc->mdsmap, mds))
  969. goto out;
  970. }
  971. /* since this file/dir wasn't known to be
  972. * replicated, then we want to look for the
  973. * authoritative mds. */
  974. if (frag.mds >= 0) {
  975. /* choose auth mds */
  976. mds = frag.mds;
  977. dout("%s %p %llx.%llx frag %u mds%d (auth)\n",
  978. __func__, inode, ceph_vinop(inode),
  979. frag.frag, mds);
  980. if (ceph_mdsmap_get_state(mdsc->mdsmap, mds) >=
  981. CEPH_MDS_STATE_ACTIVE) {
  982. if (!ceph_mdsmap_is_laggy(mdsc->mdsmap,
  983. mds))
  984. goto out;
  985. }
  986. }
  987. mode = USE_AUTH_MDS;
  988. }
  989. }
  990. spin_lock(&ci->i_ceph_lock);
  991. cap = NULL;
  992. if (mode == USE_AUTH_MDS)
  993. cap = ci->i_auth_cap;
  994. if (!cap && !RB_EMPTY_ROOT(&ci->i_caps))
  995. cap = rb_entry(rb_first(&ci->i_caps), struct ceph_cap, ci_node);
  996. if (!cap) {
  997. spin_unlock(&ci->i_ceph_lock);
  998. ceph_async_iput(inode);
  999. goto random;
  1000. }
  1001. mds = cap->session->s_mds;
  1002. dout("%s %p %llx.%llx mds%d (%scap %p)\n", __func__,
  1003. inode, ceph_vinop(inode), mds,
  1004. cap == ci->i_auth_cap ? "auth " : "", cap);
  1005. spin_unlock(&ci->i_ceph_lock);
  1006. out:
  1007. /* avoid calling iput_final() while holding mdsc->mutex or
  1008. * in mds dispatch threads */
  1009. ceph_async_iput(inode);
  1010. return mds;
  1011. random:
  1012. if (random)
  1013. *random = true;
  1014. mds = ceph_mdsmap_get_random_mds(mdsc->mdsmap);
  1015. dout("%s chose random mds%d\n", __func__, mds);
  1016. return mds;
  1017. }
  1018. /*
  1019. * session messages
  1020. */
  1021. static struct ceph_msg *create_session_msg(u32 op, u64 seq)
  1022. {
  1023. struct ceph_msg *msg;
  1024. struct ceph_mds_session_head *h;
  1025. msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h), GFP_NOFS,
  1026. false);
  1027. if (!msg) {
  1028. pr_err("create_session_msg ENOMEM creating msg\n");
  1029. return NULL;
  1030. }
  1031. h = msg->front.iov_base;
  1032. h->op = cpu_to_le32(op);
  1033. h->seq = cpu_to_le64(seq);
  1034. return msg;
  1035. }
  1036. static const unsigned char feature_bits[] = CEPHFS_FEATURES_CLIENT_SUPPORTED;
  1037. #define FEATURE_BYTES(c) (DIV_ROUND_UP((size_t)feature_bits[c - 1] + 1, 64) * 8)
  1038. static int encode_supported_features(void **p, void *end)
  1039. {
  1040. static const size_t count = ARRAY_SIZE(feature_bits);
  1041. if (count > 0) {
  1042. size_t i;
  1043. size_t size = FEATURE_BYTES(count);
  1044. if (WARN_ON_ONCE(*p + 4 + size > end))
  1045. return -ERANGE;
  1046. ceph_encode_32(p, size);
  1047. memset(*p, 0, size);
  1048. for (i = 0; i < count; i++)
  1049. ((unsigned char*)(*p))[i / 8] |= BIT(feature_bits[i] % 8);
  1050. *p += size;
  1051. } else {
  1052. if (WARN_ON_ONCE(*p + 4 > end))
  1053. return -ERANGE;
  1054. ceph_encode_32(p, 0);
  1055. }
  1056. return 0;
  1057. }
  1058. static const unsigned char metric_bits[] = CEPHFS_METRIC_SPEC_CLIENT_SUPPORTED;
  1059. #define METRIC_BYTES(cnt) (DIV_ROUND_UP((size_t)metric_bits[cnt - 1] + 1, 64) * 8)
  1060. static int encode_metric_spec(void **p, void *end)
  1061. {
  1062. static const size_t count = ARRAY_SIZE(metric_bits);
  1063. /* header */
  1064. if (WARN_ON_ONCE(*p + 2 > end))
  1065. return -ERANGE;
  1066. ceph_encode_8(p, 1); /* version */
  1067. ceph_encode_8(p, 1); /* compat */
  1068. if (count > 0) {
  1069. size_t i;
  1070. size_t size = METRIC_BYTES(count);
  1071. if (WARN_ON_ONCE(*p + 4 + 4 + size > end))
  1072. return -ERANGE;
  1073. /* metric spec info length */
  1074. ceph_encode_32(p, 4 + size);
  1075. /* metric spec */
  1076. ceph_encode_32(p, size);
  1077. memset(*p, 0, size);
  1078. for (i = 0; i < count; i++)
  1079. ((unsigned char *)(*p))[i / 8] |= BIT(metric_bits[i] % 8);
  1080. *p += size;
  1081. } else {
  1082. if (WARN_ON_ONCE(*p + 4 + 4 > end))
  1083. return -ERANGE;
  1084. /* metric spec info length */
  1085. ceph_encode_32(p, 4);
  1086. /* metric spec */
  1087. ceph_encode_32(p, 0);
  1088. }
  1089. return 0;
  1090. }
  1091. /*
  1092. * session message, specialization for CEPH_SESSION_REQUEST_OPEN
  1093. * to include additional client metadata fields.
  1094. */
  1095. static struct ceph_msg *create_session_open_msg(struct ceph_mds_client *mdsc, u64 seq)
  1096. {
  1097. struct ceph_msg *msg;
  1098. struct ceph_mds_session_head *h;
  1099. int i = -1;
  1100. int extra_bytes = 0;
  1101. int metadata_key_count = 0;
  1102. struct ceph_options *opt = mdsc->fsc->client->options;
  1103. struct ceph_mount_options *fsopt = mdsc->fsc->mount_options;
  1104. size_t size, count;
  1105. void *p, *end;
  1106. int ret;
  1107. const char* metadata[][2] = {
  1108. {"hostname", mdsc->nodename},
  1109. {"kernel_version", init_utsname()->release},
  1110. {"entity_id", opt->name ? : ""},
  1111. {"root", fsopt->server_path ? : "/"},
  1112. {NULL, NULL}
  1113. };
  1114. /* Calculate serialized length of metadata */
  1115. extra_bytes = 4; /* map length */
  1116. for (i = 0; metadata[i][0]; ++i) {
  1117. extra_bytes += 8 + strlen(metadata[i][0]) +
  1118. strlen(metadata[i][1]);
  1119. metadata_key_count++;
  1120. }
  1121. /* supported feature */
  1122. size = 0;
  1123. count = ARRAY_SIZE(feature_bits);
  1124. if (count > 0)
  1125. size = FEATURE_BYTES(count);
  1126. extra_bytes += 4 + size;
  1127. /* metric spec */
  1128. size = 0;
  1129. count = ARRAY_SIZE(metric_bits);
  1130. if (count > 0)
  1131. size = METRIC_BYTES(count);
  1132. extra_bytes += 2 + 4 + 4 + size;
  1133. /* Allocate the message */
  1134. msg = ceph_msg_new(CEPH_MSG_CLIENT_SESSION, sizeof(*h) + extra_bytes,
  1135. GFP_NOFS, false);
  1136. if (!msg) {
  1137. pr_err("create_session_msg ENOMEM creating msg\n");
  1138. return ERR_PTR(-ENOMEM);
  1139. }
  1140. p = msg->front.iov_base;
  1141. end = p + msg->front.iov_len;
  1142. h = p;
  1143. h->op = cpu_to_le32(CEPH_SESSION_REQUEST_OPEN);
  1144. h->seq = cpu_to_le64(seq);
  1145. /*
  1146. * Serialize client metadata into waiting buffer space, using
  1147. * the format that userspace expects for map<string, string>
  1148. *
  1149. * ClientSession messages with metadata are v4
  1150. */
  1151. msg->hdr.version = cpu_to_le16(4);
  1152. msg->hdr.compat_version = cpu_to_le16(1);
  1153. /* The write pointer, following the session_head structure */
  1154. p += sizeof(*h);
  1155. /* Number of entries in the map */
  1156. ceph_encode_32(&p, metadata_key_count);
  1157. /* Two length-prefixed strings for each entry in the map */
  1158. for (i = 0; metadata[i][0]; ++i) {
  1159. size_t const key_len = strlen(metadata[i][0]);
  1160. size_t const val_len = strlen(metadata[i][1]);
  1161. ceph_encode_32(&p, key_len);
  1162. memcpy(p, metadata[i][0], key_len);
  1163. p += key_len;
  1164. ceph_encode_32(&p, val_len);
  1165. memcpy(p, metadata[i][1], val_len);
  1166. p += val_len;
  1167. }
  1168. ret = encode_supported_features(&p, end);
  1169. if (ret) {
  1170. pr_err("encode_supported_features failed!\n");
  1171. ceph_msg_put(msg);
  1172. return ERR_PTR(ret);
  1173. }
  1174. ret = encode_metric_spec(&p, end);
  1175. if (ret) {
  1176. pr_err("encode_metric_spec failed!\n");
  1177. ceph_msg_put(msg);
  1178. return ERR_PTR(ret);
  1179. }
  1180. msg->front.iov_len = p - msg->front.iov_base;
  1181. msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
  1182. return msg;
  1183. }
  1184. /*
  1185. * send session open request.
  1186. *
  1187. * called under mdsc->mutex
  1188. */
  1189. static int __open_session(struct ceph_mds_client *mdsc,
  1190. struct ceph_mds_session *session)
  1191. {
  1192. struct ceph_msg *msg;
  1193. int mstate;
  1194. int mds = session->s_mds;
  1195. /* wait for mds to go active? */
  1196. mstate = ceph_mdsmap_get_state(mdsc->mdsmap, mds);
  1197. dout("open_session to mds%d (%s)\n", mds,
  1198. ceph_mds_state_name(mstate));
  1199. session->s_state = CEPH_MDS_SESSION_OPENING;
  1200. session->s_renew_requested = jiffies;
  1201. /* send connect message */
  1202. msg = create_session_open_msg(mdsc, session->s_seq);
  1203. if (IS_ERR(msg))
  1204. return PTR_ERR(msg);
  1205. ceph_con_send(&session->s_con, msg);
  1206. return 0;
  1207. }
  1208. /*
  1209. * open sessions for any export targets for the given mds
  1210. *
  1211. * called under mdsc->mutex
  1212. */
  1213. static struct ceph_mds_session *
  1214. __open_export_target_session(struct ceph_mds_client *mdsc, int target)
  1215. {
  1216. struct ceph_mds_session *session;
  1217. int ret;
  1218. session = __ceph_lookup_mds_session(mdsc, target);
  1219. if (!session) {
  1220. session = register_session(mdsc, target);
  1221. if (IS_ERR(session))
  1222. return session;
  1223. }
  1224. if (session->s_state == CEPH_MDS_SESSION_NEW ||
  1225. session->s_state == CEPH_MDS_SESSION_CLOSING) {
  1226. ret = __open_session(mdsc, session);
  1227. if (ret)
  1228. return ERR_PTR(ret);
  1229. }
  1230. return session;
  1231. }
  1232. struct ceph_mds_session *
  1233. ceph_mdsc_open_export_target_session(struct ceph_mds_client *mdsc, int target)
  1234. {
  1235. struct ceph_mds_session *session;
  1236. dout("open_export_target_session to mds%d\n", target);
  1237. mutex_lock(&mdsc->mutex);
  1238. session = __open_export_target_session(mdsc, target);
  1239. mutex_unlock(&mdsc->mutex);
  1240. return session;
  1241. }
  1242. static void __open_export_target_sessions(struct ceph_mds_client *mdsc,
  1243. struct ceph_mds_session *session)
  1244. {
  1245. struct ceph_mds_info *mi;
  1246. struct ceph_mds_session *ts;
  1247. int i, mds = session->s_mds;
  1248. if (mds >= mdsc->mdsmap->possible_max_rank)
  1249. return;
  1250. mi = &mdsc->mdsmap->m_info[mds];
  1251. dout("open_export_target_sessions for mds%d (%d targets)\n",
  1252. session->s_mds, mi->num_export_targets);
  1253. for (i = 0; i < mi->num_export_targets; i++) {
  1254. ts = __open_export_target_session(mdsc, mi->export_targets[i]);
  1255. ceph_put_mds_session(ts);
  1256. }
  1257. }
  1258. void ceph_mdsc_open_export_target_sessions(struct ceph_mds_client *mdsc,
  1259. struct ceph_mds_session *session)
  1260. {
  1261. mutex_lock(&mdsc->mutex);
  1262. __open_export_target_sessions(mdsc, session);
  1263. mutex_unlock(&mdsc->mutex);
  1264. }
  1265. /*
  1266. * session caps
  1267. */
  1268. static void detach_cap_releases(struct ceph_mds_session *session,
  1269. struct list_head *target)
  1270. {
  1271. lockdep_assert_held(&session->s_cap_lock);
  1272. list_splice_init(&session->s_cap_releases, target);
  1273. session->s_num_cap_releases = 0;
  1274. dout("dispose_cap_releases mds%d\n", session->s_mds);
  1275. }
  1276. static void dispose_cap_releases(struct ceph_mds_client *mdsc,
  1277. struct list_head *dispose)
  1278. {
  1279. while (!list_empty(dispose)) {
  1280. struct ceph_cap *cap;
  1281. /* zero out the in-progress message */
  1282. cap = list_first_entry(dispose, struct ceph_cap, session_caps);
  1283. list_del(&cap->session_caps);
  1284. ceph_put_cap(mdsc, cap);
  1285. }
  1286. }
  1287. static void cleanup_session_requests(struct ceph_mds_client *mdsc,
  1288. struct ceph_mds_session *session)
  1289. {
  1290. struct ceph_mds_request *req;
  1291. struct rb_node *p;
  1292. dout("cleanup_session_requests mds%d\n", session->s_mds);
  1293. mutex_lock(&mdsc->mutex);
  1294. while (!list_empty(&session->s_unsafe)) {
  1295. req = list_first_entry(&session->s_unsafe,
  1296. struct ceph_mds_request, r_unsafe_item);
  1297. pr_warn_ratelimited(" dropping unsafe request %llu\n",
  1298. req->r_tid);
  1299. if (req->r_target_inode)
  1300. mapping_set_error(req->r_target_inode->i_mapping, -EIO);
  1301. if (req->r_unsafe_dir)
  1302. mapping_set_error(req->r_unsafe_dir->i_mapping, -EIO);
  1303. __unregister_request(mdsc, req);
  1304. }
  1305. /* zero r_attempts, so kick_requests() will re-send requests */
  1306. p = rb_first(&mdsc->request_tree);
  1307. while (p) {
  1308. req = rb_entry(p, struct ceph_mds_request, r_node);
  1309. p = rb_next(p);
  1310. if (req->r_session &&
  1311. req->r_session->s_mds == session->s_mds)
  1312. req->r_attempts = 0;
  1313. }
  1314. mutex_unlock(&mdsc->mutex);
  1315. }
  1316. /*
  1317. * Helper to safely iterate over all caps associated with a session, with
  1318. * special care taken to handle a racing __ceph_remove_cap().
  1319. *
  1320. * Caller must hold session s_mutex.
  1321. */
  1322. int ceph_iterate_session_caps(struct ceph_mds_session *session,
  1323. int (*cb)(struct inode *, struct ceph_cap *,
  1324. void *), void *arg)
  1325. {
  1326. struct list_head *p;
  1327. struct ceph_cap *cap;
  1328. struct inode *inode, *last_inode = NULL;
  1329. struct ceph_cap *old_cap = NULL;
  1330. int ret;
  1331. dout("iterate_session_caps %p mds%d\n", session, session->s_mds);
  1332. spin_lock(&session->s_cap_lock);
  1333. p = session->s_caps.next;
  1334. while (p != &session->s_caps) {
  1335. cap = list_entry(p, struct ceph_cap, session_caps);
  1336. inode = igrab(&cap->ci->vfs_inode);
  1337. if (!inode) {
  1338. p = p->next;
  1339. continue;
  1340. }
  1341. session->s_cap_iterator = cap;
  1342. spin_unlock(&session->s_cap_lock);
  1343. if (last_inode) {
  1344. /* avoid calling iput_final() while holding
  1345. * s_mutex or in mds dispatch threads */
  1346. ceph_async_iput(last_inode);
  1347. last_inode = NULL;
  1348. }
  1349. if (old_cap) {
  1350. ceph_put_cap(session->s_mdsc, old_cap);
  1351. old_cap = NULL;
  1352. }
  1353. ret = cb(inode, cap, arg);
  1354. last_inode = inode;
  1355. spin_lock(&session->s_cap_lock);
  1356. p = p->next;
  1357. if (!cap->ci) {
  1358. dout("iterate_session_caps finishing cap %p removal\n",
  1359. cap);
  1360. BUG_ON(cap->session != session);
  1361. cap->session = NULL;
  1362. list_del_init(&cap->session_caps);
  1363. session->s_nr_caps--;
  1364. atomic64_dec(&session->s_mdsc->metric.total_caps);
  1365. if (cap->queue_release)
  1366. __ceph_queue_cap_release(session, cap);
  1367. else
  1368. old_cap = cap; /* put_cap it w/o locks held */
  1369. }
  1370. if (ret < 0)
  1371. goto out;
  1372. }
  1373. ret = 0;
  1374. out:
  1375. session->s_cap_iterator = NULL;
  1376. spin_unlock(&session->s_cap_lock);
  1377. ceph_async_iput(last_inode);
  1378. if (old_cap)
  1379. ceph_put_cap(session->s_mdsc, old_cap);
  1380. return ret;
  1381. }
  1382. static int remove_capsnaps(struct ceph_mds_client *mdsc, struct inode *inode)
  1383. {
  1384. struct ceph_inode_info *ci = ceph_inode(inode);
  1385. struct ceph_cap_snap *capsnap;
  1386. int capsnap_release = 0;
  1387. lockdep_assert_held(&ci->i_ceph_lock);
  1388. dout("removing capsnaps, ci is %p, inode is %p\n", ci, inode);
  1389. while (!list_empty(&ci->i_cap_snaps)) {
  1390. capsnap = list_first_entry(&ci->i_cap_snaps,
  1391. struct ceph_cap_snap, ci_item);
  1392. __ceph_remove_capsnap(inode, capsnap, NULL, NULL);
  1393. ceph_put_snap_context(capsnap->context);
  1394. ceph_put_cap_snap(capsnap);
  1395. capsnap_release++;
  1396. }
  1397. wake_up_all(&ci->i_cap_wq);
  1398. wake_up_all(&mdsc->cap_flushing_wq);
  1399. return capsnap_release;
  1400. }
  1401. static int remove_session_caps_cb(struct inode *inode, struct ceph_cap *cap,
  1402. void *arg)
  1403. {
  1404. struct ceph_fs_client *fsc = (struct ceph_fs_client *)arg;
  1405. struct ceph_mds_client *mdsc = fsc->mdsc;
  1406. struct ceph_inode_info *ci = ceph_inode(inode);
  1407. LIST_HEAD(to_remove);
  1408. bool dirty_dropped = false;
  1409. bool invalidate = false;
  1410. int capsnap_release = 0;
  1411. dout("removing cap %p, ci is %p, inode is %p\n",
  1412. cap, ci, &ci->vfs_inode);
  1413. spin_lock(&ci->i_ceph_lock);
  1414. __ceph_remove_cap(cap, false);
  1415. if (!ci->i_auth_cap) {
  1416. struct ceph_cap_flush *cf;
  1417. if (READ_ONCE(fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) {
  1418. if (inode->i_data.nrpages > 0)
  1419. invalidate = true;
  1420. if (ci->i_wrbuffer_ref > 0)
  1421. mapping_set_error(&inode->i_data, -EIO);
  1422. }
  1423. while (!list_empty(&ci->i_cap_flush_list)) {
  1424. cf = list_first_entry(&ci->i_cap_flush_list,
  1425. struct ceph_cap_flush, i_list);
  1426. list_move(&cf->i_list, &to_remove);
  1427. }
  1428. spin_lock(&mdsc->cap_dirty_lock);
  1429. list_for_each_entry(cf, &to_remove, i_list)
  1430. list_del_init(&cf->g_list);
  1431. if (!list_empty(&ci->i_dirty_item)) {
  1432. pr_warn_ratelimited(
  1433. " dropping dirty %s state for %p %lld\n",
  1434. ceph_cap_string(ci->i_dirty_caps),
  1435. inode, ceph_ino(inode));
  1436. ci->i_dirty_caps = 0;
  1437. list_del_init(&ci->i_dirty_item);
  1438. dirty_dropped = true;
  1439. }
  1440. if (!list_empty(&ci->i_flushing_item)) {
  1441. pr_warn_ratelimited(
  1442. " dropping dirty+flushing %s state for %p %lld\n",
  1443. ceph_cap_string(ci->i_flushing_caps),
  1444. inode, ceph_ino(inode));
  1445. ci->i_flushing_caps = 0;
  1446. list_del_init(&ci->i_flushing_item);
  1447. mdsc->num_cap_flushing--;
  1448. dirty_dropped = true;
  1449. }
  1450. spin_unlock(&mdsc->cap_dirty_lock);
  1451. if (dirty_dropped) {
  1452. mapping_set_error(inode->i_mapping, -EIO);
  1453. if (ci->i_wrbuffer_ref_head == 0 &&
  1454. ci->i_wr_ref == 0 &&
  1455. ci->i_dirty_caps == 0 &&
  1456. ci->i_flushing_caps == 0) {
  1457. ceph_put_snap_context(ci->i_head_snapc);
  1458. ci->i_head_snapc = NULL;
  1459. }
  1460. }
  1461. if (atomic_read(&ci->i_filelock_ref) > 0) {
  1462. /* make further file lock syscall return -EIO */
  1463. ci->i_ceph_flags |= CEPH_I_ERROR_FILELOCK;
  1464. pr_warn_ratelimited(" dropping file locks for %p %lld\n",
  1465. inode, ceph_ino(inode));
  1466. }
  1467. if (!ci->i_dirty_caps && ci->i_prealloc_cap_flush) {
  1468. list_add(&ci->i_prealloc_cap_flush->i_list, &to_remove);
  1469. ci->i_prealloc_cap_flush = NULL;
  1470. }
  1471. if (!list_empty(&ci->i_cap_snaps))
  1472. capsnap_release = remove_capsnaps(mdsc, inode);
  1473. }
  1474. spin_unlock(&ci->i_ceph_lock);
  1475. while (!list_empty(&to_remove)) {
  1476. struct ceph_cap_flush *cf;
  1477. cf = list_first_entry(&to_remove,
  1478. struct ceph_cap_flush, i_list);
  1479. list_del_init(&cf->i_list);
  1480. if (!cf->is_capsnap)
  1481. ceph_free_cap_flush(cf);
  1482. }
  1483. wake_up_all(&ci->i_cap_wq);
  1484. if (invalidate)
  1485. ceph_queue_invalidate(inode);
  1486. if (dirty_dropped)
  1487. iput(inode);
  1488. while (capsnap_release--)
  1489. iput(inode);
  1490. return 0;
  1491. }
  1492. /*
  1493. * caller must hold session s_mutex
  1494. */
  1495. static void remove_session_caps(struct ceph_mds_session *session)
  1496. {
  1497. struct ceph_fs_client *fsc = session->s_mdsc->fsc;
  1498. struct super_block *sb = fsc->sb;
  1499. LIST_HEAD(dispose);
  1500. dout("remove_session_caps on %p\n", session);
  1501. ceph_iterate_session_caps(session, remove_session_caps_cb, fsc);
  1502. wake_up_all(&fsc->mdsc->cap_flushing_wq);
  1503. spin_lock(&session->s_cap_lock);
  1504. if (session->s_nr_caps > 0) {
  1505. struct inode *inode;
  1506. struct ceph_cap *cap, *prev = NULL;
  1507. struct ceph_vino vino;
  1508. /*
  1509. * iterate_session_caps() skips inodes that are being
  1510. * deleted, we need to wait until deletions are complete.
  1511. * __wait_on_freeing_inode() is designed for the job,
  1512. * but it is not exported, so use lookup inode function
  1513. * to access it.
  1514. */
  1515. while (!list_empty(&session->s_caps)) {
  1516. cap = list_entry(session->s_caps.next,
  1517. struct ceph_cap, session_caps);
  1518. if (cap == prev)
  1519. break;
  1520. prev = cap;
  1521. vino = cap->ci->i_vino;
  1522. spin_unlock(&session->s_cap_lock);
  1523. inode = ceph_find_inode(sb, vino);
  1524. /* avoid calling iput_final() while holding s_mutex */
  1525. ceph_async_iput(inode);
  1526. spin_lock(&session->s_cap_lock);
  1527. }
  1528. }
  1529. // drop cap expires and unlock s_cap_lock
  1530. detach_cap_releases(session, &dispose);
  1531. BUG_ON(session->s_nr_caps > 0);
  1532. BUG_ON(!list_empty(&session->s_cap_flushing));
  1533. spin_unlock(&session->s_cap_lock);
  1534. dispose_cap_releases(session->s_mdsc, &dispose);
  1535. }
  1536. enum {
  1537. RECONNECT,
  1538. RENEWCAPS,
  1539. FORCE_RO,
  1540. };
  1541. /*
  1542. * wake up any threads waiting on this session's caps. if the cap is
  1543. * old (didn't get renewed on the client reconnect), remove it now.
  1544. *
  1545. * caller must hold s_mutex.
  1546. */
  1547. static int wake_up_session_cb(struct inode *inode, struct ceph_cap *cap,
  1548. void *arg)
  1549. {
  1550. struct ceph_inode_info *ci = ceph_inode(inode);
  1551. unsigned long ev = (unsigned long)arg;
  1552. if (ev == RECONNECT) {
  1553. spin_lock(&ci->i_ceph_lock);
  1554. ci->i_wanted_max_size = 0;
  1555. ci->i_requested_max_size = 0;
  1556. spin_unlock(&ci->i_ceph_lock);
  1557. } else if (ev == RENEWCAPS) {
  1558. if (cap->cap_gen < cap->session->s_cap_gen) {
  1559. /* mds did not re-issue stale cap */
  1560. spin_lock(&ci->i_ceph_lock);
  1561. cap->issued = cap->implemented = CEPH_CAP_PIN;
  1562. spin_unlock(&ci->i_ceph_lock);
  1563. }
  1564. } else if (ev == FORCE_RO) {
  1565. }
  1566. wake_up_all(&ci->i_cap_wq);
  1567. return 0;
  1568. }
  1569. static void wake_up_session_caps(struct ceph_mds_session *session, int ev)
  1570. {
  1571. dout("wake_up_session_caps %p mds%d\n", session, session->s_mds);
  1572. ceph_iterate_session_caps(session, wake_up_session_cb,
  1573. (void *)(unsigned long)ev);
  1574. }
  1575. /*
  1576. * Send periodic message to MDS renewing all currently held caps. The
  1577. * ack will reset the expiration for all caps from this session.
  1578. *
  1579. * caller holds s_mutex
  1580. */
  1581. static int send_renew_caps(struct ceph_mds_client *mdsc,
  1582. struct ceph_mds_session *session)
  1583. {
  1584. struct ceph_msg *msg;
  1585. int state;
  1586. if (time_after_eq(jiffies, session->s_cap_ttl) &&
  1587. time_after_eq(session->s_cap_ttl, session->s_renew_requested))
  1588. pr_info("mds%d caps stale\n", session->s_mds);
  1589. session->s_renew_requested = jiffies;
  1590. /* do not try to renew caps until a recovering mds has reconnected
  1591. * with its clients. */
  1592. state = ceph_mdsmap_get_state(mdsc->mdsmap, session->s_mds);
  1593. if (state < CEPH_MDS_STATE_RECONNECT) {
  1594. dout("send_renew_caps ignoring mds%d (%s)\n",
  1595. session->s_mds, ceph_mds_state_name(state));
  1596. return 0;
  1597. }
  1598. dout("send_renew_caps to mds%d (%s)\n", session->s_mds,
  1599. ceph_mds_state_name(state));
  1600. msg = create_session_msg(CEPH_SESSION_REQUEST_RENEWCAPS,
  1601. ++session->s_renew_seq);
  1602. if (!msg)
  1603. return -ENOMEM;
  1604. ceph_con_send(&session->s_con, msg);
  1605. return 0;
  1606. }
  1607. static int send_flushmsg_ack(struct ceph_mds_client *mdsc,
  1608. struct ceph_mds_session *session, u64 seq)
  1609. {
  1610. struct ceph_msg *msg;
  1611. dout("send_flushmsg_ack to mds%d (%s)s seq %lld\n",
  1612. session->s_mds, ceph_session_state_name(session->s_state), seq);
  1613. msg = create_session_msg(CEPH_SESSION_FLUSHMSG_ACK, seq);
  1614. if (!msg)
  1615. return -ENOMEM;
  1616. ceph_con_send(&session->s_con, msg);
  1617. return 0;
  1618. }
  1619. /*
  1620. * Note new cap ttl, and any transition from stale -> not stale (fresh?).
  1621. *
  1622. * Called under session->s_mutex
  1623. */
  1624. static void renewed_caps(struct ceph_mds_client *mdsc,
  1625. struct ceph_mds_session *session, int is_renew)
  1626. {
  1627. int was_stale;
  1628. int wake = 0;
  1629. spin_lock(&session->s_cap_lock);
  1630. was_stale = is_renew && time_after_eq(jiffies, session->s_cap_ttl);
  1631. session->s_cap_ttl = session->s_renew_requested +
  1632. mdsc->mdsmap->m_session_timeout*HZ;
  1633. if (was_stale) {
  1634. if (time_before(jiffies, session->s_cap_ttl)) {
  1635. pr_info("mds%d caps renewed\n", session->s_mds);
  1636. wake = 1;
  1637. } else {
  1638. pr_info("mds%d caps still stale\n", session->s_mds);
  1639. }
  1640. }
  1641. dout("renewed_caps mds%d ttl now %lu, was %s, now %s\n",
  1642. session->s_mds, session->s_cap_ttl, was_stale ? "stale" : "fresh",
  1643. time_before(jiffies, session->s_cap_ttl) ? "stale" : "fresh");
  1644. spin_unlock(&session->s_cap_lock);
  1645. if (wake)
  1646. wake_up_session_caps(session, RENEWCAPS);
  1647. }
  1648. /*
  1649. * send a session close request
  1650. */
  1651. static int request_close_session(struct ceph_mds_session *session)
  1652. {
  1653. struct ceph_msg *msg;
  1654. dout("request_close_session mds%d state %s seq %lld\n",
  1655. session->s_mds, ceph_session_state_name(session->s_state),
  1656. session->s_seq);
  1657. msg = create_session_msg(CEPH_SESSION_REQUEST_CLOSE, session->s_seq);
  1658. if (!msg)
  1659. return -ENOMEM;
  1660. ceph_con_send(&session->s_con, msg);
  1661. return 1;
  1662. }
  1663. /*
  1664. * Called with s_mutex held.
  1665. */
  1666. static int __close_session(struct ceph_mds_client *mdsc,
  1667. struct ceph_mds_session *session)
  1668. {
  1669. if (session->s_state >= CEPH_MDS_SESSION_CLOSING)
  1670. return 0;
  1671. session->s_state = CEPH_MDS_SESSION_CLOSING;
  1672. return request_close_session(session);
  1673. }
  1674. static bool drop_negative_children(struct dentry *dentry)
  1675. {
  1676. struct dentry *child;
  1677. bool all_negative = true;
  1678. if (!d_is_dir(dentry))
  1679. goto out;
  1680. spin_lock(&dentry->d_lock);
  1681. list_for_each_entry(child, &dentry->d_subdirs, d_child) {
  1682. if (d_really_is_positive(child)) {
  1683. all_negative = false;
  1684. break;
  1685. }
  1686. }
  1687. spin_unlock(&dentry->d_lock);
  1688. if (all_negative)
  1689. shrink_dcache_parent(dentry);
  1690. out:
  1691. return all_negative;
  1692. }
  1693. /*
  1694. * Trim old(er) caps.
  1695. *
  1696. * Because we can't cache an inode without one or more caps, we do
  1697. * this indirectly: if a cap is unused, we prune its aliases, at which
  1698. * point the inode will hopefully get dropped to.
  1699. *
  1700. * Yes, this is a bit sloppy. Our only real goal here is to respond to
  1701. * memory pressure from the MDS, though, so it needn't be perfect.
  1702. */
  1703. static int trim_caps_cb(struct inode *inode, struct ceph_cap *cap, void *arg)
  1704. {
  1705. int *remaining = arg;
  1706. struct ceph_inode_info *ci = ceph_inode(inode);
  1707. int used, wanted, oissued, mine;
  1708. if (*remaining <= 0)
  1709. return -1;
  1710. spin_lock(&ci->i_ceph_lock);
  1711. mine = cap->issued | cap->implemented;
  1712. used = __ceph_caps_used(ci);
  1713. wanted = __ceph_caps_file_wanted(ci);
  1714. oissued = __ceph_caps_issued_other(ci, cap);
  1715. dout("trim_caps_cb %p cap %p mine %s oissued %s used %s wanted %s\n",
  1716. inode, cap, ceph_cap_string(mine), ceph_cap_string(oissued),
  1717. ceph_cap_string(used), ceph_cap_string(wanted));
  1718. if (cap == ci->i_auth_cap) {
  1719. if (ci->i_dirty_caps || ci->i_flushing_caps ||
  1720. !list_empty(&ci->i_cap_snaps))
  1721. goto out;
  1722. if ((used | wanted) & CEPH_CAP_ANY_WR)
  1723. goto out;
  1724. /* Note: it's possible that i_filelock_ref becomes non-zero
  1725. * after dropping auth caps. It doesn't hurt because reply
  1726. * of lock mds request will re-add auth caps. */
  1727. if (atomic_read(&ci->i_filelock_ref) > 0)
  1728. goto out;
  1729. }
  1730. /* The inode has cached pages, but it's no longer used.
  1731. * we can safely drop it */
  1732. if (S_ISREG(inode->i_mode) &&
  1733. wanted == 0 && used == CEPH_CAP_FILE_CACHE &&
  1734. !(oissued & CEPH_CAP_FILE_CACHE)) {
  1735. used = 0;
  1736. oissued = 0;
  1737. }
  1738. if ((used | wanted) & ~oissued & mine)
  1739. goto out; /* we need these caps */
  1740. if (oissued) {
  1741. /* we aren't the only cap.. just remove us */
  1742. __ceph_remove_cap(cap, true);
  1743. (*remaining)--;
  1744. } else {
  1745. struct dentry *dentry;
  1746. /* try dropping referring dentries */
  1747. spin_unlock(&ci->i_ceph_lock);
  1748. dentry = d_find_any_alias(inode);
  1749. if (dentry && drop_negative_children(dentry)) {
  1750. int count;
  1751. dput(dentry);
  1752. d_prune_aliases(inode);
  1753. count = atomic_read(&inode->i_count);
  1754. if (count == 1)
  1755. (*remaining)--;
  1756. dout("trim_caps_cb %p cap %p pruned, count now %d\n",
  1757. inode, cap, count);
  1758. } else {
  1759. dput(dentry);
  1760. }
  1761. return 0;
  1762. }
  1763. out:
  1764. spin_unlock(&ci->i_ceph_lock);
  1765. return 0;
  1766. }
  1767. /*
  1768. * Trim session cap count down to some max number.
  1769. */
  1770. int ceph_trim_caps(struct ceph_mds_client *mdsc,
  1771. struct ceph_mds_session *session,
  1772. int max_caps)
  1773. {
  1774. int trim_caps = session->s_nr_caps - max_caps;
  1775. dout("trim_caps mds%d start: %d / %d, trim %d\n",
  1776. session->s_mds, session->s_nr_caps, max_caps, trim_caps);
  1777. if (trim_caps > 0) {
  1778. int remaining = trim_caps;
  1779. ceph_iterate_session_caps(session, trim_caps_cb, &remaining);
  1780. dout("trim_caps mds%d done: %d / %d, trimmed %d\n",
  1781. session->s_mds, session->s_nr_caps, max_caps,
  1782. trim_caps - remaining);
  1783. }
  1784. ceph_flush_cap_releases(mdsc, session);
  1785. return 0;
  1786. }
  1787. static int check_caps_flush(struct ceph_mds_client *mdsc,
  1788. u64 want_flush_tid)
  1789. {
  1790. int ret = 1;
  1791. spin_lock(&mdsc->cap_dirty_lock);
  1792. if (!list_empty(&mdsc->cap_flush_list)) {
  1793. struct ceph_cap_flush *cf =
  1794. list_first_entry(&mdsc->cap_flush_list,
  1795. struct ceph_cap_flush, g_list);
  1796. if (cf->tid <= want_flush_tid) {
  1797. dout("check_caps_flush still flushing tid "
  1798. "%llu <= %llu\n", cf->tid, want_flush_tid);
  1799. ret = 0;
  1800. }
  1801. }
  1802. spin_unlock(&mdsc->cap_dirty_lock);
  1803. return ret;
  1804. }
  1805. /*
  1806. * flush all dirty inode data to disk.
  1807. *
  1808. * returns true if we've flushed through want_flush_tid
  1809. */
  1810. static void wait_caps_flush(struct ceph_mds_client *mdsc,
  1811. u64 want_flush_tid)
  1812. {
  1813. dout("check_caps_flush want %llu\n", want_flush_tid);
  1814. wait_event(mdsc->cap_flushing_wq,
  1815. check_caps_flush(mdsc, want_flush_tid));
  1816. dout("check_caps_flush ok, flushed thru %llu\n", want_flush_tid);
  1817. }
  1818. /*
  1819. * called under s_mutex
  1820. */
  1821. static void ceph_send_cap_releases(struct ceph_mds_client *mdsc,
  1822. struct ceph_mds_session *session)
  1823. {
  1824. struct ceph_msg *msg = NULL;
  1825. struct ceph_mds_cap_release *head;
  1826. struct ceph_mds_cap_item *item;
  1827. struct ceph_osd_client *osdc = &mdsc->fsc->client->osdc;
  1828. struct ceph_cap *cap;
  1829. LIST_HEAD(tmp_list);
  1830. int num_cap_releases;
  1831. __le32 barrier, *cap_barrier;
  1832. down_read(&osdc->lock);
  1833. barrier = cpu_to_le32(osdc->epoch_barrier);
  1834. up_read(&osdc->lock);
  1835. spin_lock(&session->s_cap_lock);
  1836. again:
  1837. list_splice_init(&session->s_cap_releases, &tmp_list);
  1838. num_cap_releases = session->s_num_cap_releases;
  1839. session->s_num_cap_releases = 0;
  1840. spin_unlock(&session->s_cap_lock);
  1841. while (!list_empty(&tmp_list)) {
  1842. if (!msg) {
  1843. msg = ceph_msg_new(CEPH_MSG_CLIENT_CAPRELEASE,
  1844. PAGE_SIZE, GFP_NOFS, false);
  1845. if (!msg)
  1846. goto out_err;
  1847. head = msg->front.iov_base;
  1848. head->num = cpu_to_le32(0);
  1849. msg->front.iov_len = sizeof(*head);
  1850. msg->hdr.version = cpu_to_le16(2);
  1851. msg->hdr.compat_version = cpu_to_le16(1);
  1852. }
  1853. cap = list_first_entry(&tmp_list, struct ceph_cap,
  1854. session_caps);
  1855. list_del(&cap->session_caps);
  1856. num_cap_releases--;
  1857. head = msg->front.iov_base;
  1858. put_unaligned_le32(get_unaligned_le32(&head->num) + 1,
  1859. &head->num);
  1860. item = msg->front.iov_base + msg->front.iov_len;
  1861. item->ino = cpu_to_le64(cap->cap_ino);
  1862. item->cap_id = cpu_to_le64(cap->cap_id);
  1863. item->migrate_seq = cpu_to_le32(cap->mseq);
  1864. item->seq = cpu_to_le32(cap->issue_seq);
  1865. msg->front.iov_len += sizeof(*item);
  1866. ceph_put_cap(mdsc, cap);
  1867. if (le32_to_cpu(head->num) == CEPH_CAPS_PER_RELEASE) {
  1868. // Append cap_barrier field
  1869. cap_barrier = msg->front.iov_base + msg->front.iov_len;
  1870. *cap_barrier = barrier;
  1871. msg->front.iov_len += sizeof(*cap_barrier);
  1872. msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
  1873. dout("send_cap_releases mds%d %p\n", session->s_mds, msg);
  1874. ceph_con_send(&session->s_con, msg);
  1875. msg = NULL;
  1876. }
  1877. }
  1878. BUG_ON(num_cap_releases != 0);
  1879. spin_lock(&session->s_cap_lock);
  1880. if (!list_empty(&session->s_cap_releases))
  1881. goto again;
  1882. spin_unlock(&session->s_cap_lock);
  1883. if (msg) {
  1884. // Append cap_barrier field
  1885. cap_barrier = msg->front.iov_base + msg->front.iov_len;
  1886. *cap_barrier = barrier;
  1887. msg->front.iov_len += sizeof(*cap_barrier);
  1888. msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
  1889. dout("send_cap_releases mds%d %p\n", session->s_mds, msg);
  1890. ceph_con_send(&session->s_con, msg);
  1891. }
  1892. return;
  1893. out_err:
  1894. pr_err("send_cap_releases mds%d, failed to allocate message\n",
  1895. session->s_mds);
  1896. spin_lock(&session->s_cap_lock);
  1897. list_splice(&tmp_list, &session->s_cap_releases);
  1898. session->s_num_cap_releases += num_cap_releases;
  1899. spin_unlock(&session->s_cap_lock);
  1900. }
  1901. static void ceph_cap_release_work(struct work_struct *work)
  1902. {
  1903. struct ceph_mds_session *session =
  1904. container_of(work, struct ceph_mds_session, s_cap_release_work);
  1905. mutex_lock(&session->s_mutex);
  1906. if (session->s_state == CEPH_MDS_SESSION_OPEN ||
  1907. session->s_state == CEPH_MDS_SESSION_HUNG)
  1908. ceph_send_cap_releases(session->s_mdsc, session);
  1909. mutex_unlock(&session->s_mutex);
  1910. ceph_put_mds_session(session);
  1911. }
  1912. void ceph_flush_cap_releases(struct ceph_mds_client *mdsc,
  1913. struct ceph_mds_session *session)
  1914. {
  1915. if (mdsc->stopping)
  1916. return;
  1917. ceph_get_mds_session(session);
  1918. if (queue_work(mdsc->fsc->cap_wq,
  1919. &session->s_cap_release_work)) {
  1920. dout("cap release work queued\n");
  1921. } else {
  1922. ceph_put_mds_session(session);
  1923. dout("failed to queue cap release work\n");
  1924. }
  1925. }
  1926. /*
  1927. * caller holds session->s_cap_lock
  1928. */
  1929. void __ceph_queue_cap_release(struct ceph_mds_session *session,
  1930. struct ceph_cap *cap)
  1931. {
  1932. list_add_tail(&cap->session_caps, &session->s_cap_releases);
  1933. session->s_num_cap_releases++;
  1934. if (!(session->s_num_cap_releases % CEPH_CAPS_PER_RELEASE))
  1935. ceph_flush_cap_releases(session->s_mdsc, session);
  1936. }
  1937. static void ceph_cap_reclaim_work(struct work_struct *work)
  1938. {
  1939. struct ceph_mds_client *mdsc =
  1940. container_of(work, struct ceph_mds_client, cap_reclaim_work);
  1941. int ret = ceph_trim_dentries(mdsc);
  1942. if (ret == -EAGAIN)
  1943. ceph_queue_cap_reclaim_work(mdsc);
  1944. }
  1945. void ceph_queue_cap_reclaim_work(struct ceph_mds_client *mdsc)
  1946. {
  1947. if (mdsc->stopping)
  1948. return;
  1949. if (queue_work(mdsc->fsc->cap_wq, &mdsc->cap_reclaim_work)) {
  1950. dout("caps reclaim work queued\n");
  1951. } else {
  1952. dout("failed to queue caps release work\n");
  1953. }
  1954. }
  1955. void ceph_reclaim_caps_nr(struct ceph_mds_client *mdsc, int nr)
  1956. {
  1957. int val;
  1958. if (!nr)
  1959. return;
  1960. val = atomic_add_return(nr, &mdsc->cap_reclaim_pending);
  1961. if ((val % CEPH_CAPS_PER_RELEASE) < nr) {
  1962. atomic_set(&mdsc->cap_reclaim_pending, 0);
  1963. ceph_queue_cap_reclaim_work(mdsc);
  1964. }
  1965. }
  1966. /*
  1967. * requests
  1968. */
  1969. int ceph_alloc_readdir_reply_buffer(struct ceph_mds_request *req,
  1970. struct inode *dir)
  1971. {
  1972. struct ceph_inode_info *ci = ceph_inode(dir);
  1973. struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;
  1974. struct ceph_mount_options *opt = req->r_mdsc->fsc->mount_options;
  1975. size_t size = sizeof(struct ceph_mds_reply_dir_entry);
  1976. unsigned int num_entries;
  1977. int order;
  1978. spin_lock(&ci->i_ceph_lock);
  1979. num_entries = ci->i_files + ci->i_subdirs;
  1980. spin_unlock(&ci->i_ceph_lock);
  1981. num_entries = max(num_entries, 1U);
  1982. num_entries = min(num_entries, opt->max_readdir);
  1983. order = get_order(size * num_entries);
  1984. while (order >= 0) {
  1985. rinfo->dir_entries = (void*)__get_free_pages(GFP_KERNEL |
  1986. __GFP_NOWARN,
  1987. order);
  1988. if (rinfo->dir_entries)
  1989. break;
  1990. order--;
  1991. }
  1992. if (!rinfo->dir_entries)
  1993. return -ENOMEM;
  1994. num_entries = (PAGE_SIZE << order) / size;
  1995. num_entries = min(num_entries, opt->max_readdir);
  1996. rinfo->dir_buf_size = PAGE_SIZE << order;
  1997. req->r_num_caps = num_entries + 1;
  1998. req->r_args.readdir.max_entries = cpu_to_le32(num_entries);
  1999. req->r_args.readdir.max_bytes = cpu_to_le32(opt->max_readdir_bytes);
  2000. return 0;
  2001. }
  2002. /*
  2003. * Create an mds request.
  2004. */
  2005. struct ceph_mds_request *
  2006. ceph_mdsc_create_request(struct ceph_mds_client *mdsc, int op, int mode)
  2007. {
  2008. struct ceph_mds_request *req;
  2009. req = kmem_cache_zalloc(ceph_mds_request_cachep, GFP_NOFS);
  2010. if (!req)
  2011. return ERR_PTR(-ENOMEM);
  2012. mutex_init(&req->r_fill_mutex);
  2013. req->r_mdsc = mdsc;
  2014. req->r_started = jiffies;
  2015. req->r_start_latency = ktime_get();
  2016. req->r_resend_mds = -1;
  2017. INIT_LIST_HEAD(&req->r_unsafe_dir_item);
  2018. INIT_LIST_HEAD(&req->r_unsafe_target_item);
  2019. req->r_fmode = -1;
  2020. kref_init(&req->r_kref);
  2021. RB_CLEAR_NODE(&req->r_node);
  2022. INIT_LIST_HEAD(&req->r_wait);
  2023. init_completion(&req->r_completion);
  2024. init_completion(&req->r_safe_completion);
  2025. INIT_LIST_HEAD(&req->r_unsafe_item);
  2026. ktime_get_coarse_real_ts64(&req->r_stamp);
  2027. req->r_op = op;
  2028. req->r_direct_mode = mode;
  2029. return req;
  2030. }
  2031. /*
  2032. * return oldest (lowest) request, tid in request tree, 0 if none.
  2033. *
  2034. * called under mdsc->mutex.
  2035. */
  2036. static struct ceph_mds_request *__get_oldest_req(struct ceph_mds_client *mdsc)
  2037. {
  2038. if (RB_EMPTY_ROOT(&mdsc->request_tree))
  2039. return NULL;
  2040. return rb_entry(rb_first(&mdsc->request_tree),
  2041. struct ceph_mds_request, r_node);
  2042. }
  2043. static inline u64 __get_oldest_tid(struct ceph_mds_client *mdsc)
  2044. {
  2045. return mdsc->oldest_tid;
  2046. }
  2047. /*
  2048. * Build a dentry's path. Allocate on heap; caller must kfree. Based
  2049. * on build_path_from_dentry in fs/cifs/dir.c.
  2050. *
  2051. * If @stop_on_nosnap, generate path relative to the first non-snapped
  2052. * inode.
  2053. *
  2054. * Encode hidden .snap dirs as a double /, i.e.
  2055. * foo/.snap/bar -> foo//bar
  2056. */
  2057. char *ceph_mdsc_build_path(struct dentry *dentry, int *plen, u64 *pbase,
  2058. int stop_on_nosnap)
  2059. {
  2060. struct dentry *temp;
  2061. char *path;
  2062. int pos;
  2063. unsigned seq;
  2064. u64 base;
  2065. if (!dentry)
  2066. return ERR_PTR(-EINVAL);
  2067. path = __getname();
  2068. if (!path)
  2069. return ERR_PTR(-ENOMEM);
  2070. retry:
  2071. pos = PATH_MAX - 1;
  2072. path[pos] = '\0';
  2073. seq = read_seqbegin(&rename_lock);
  2074. rcu_read_lock();
  2075. temp = dentry;
  2076. for (;;) {
  2077. struct inode *inode;
  2078. spin_lock(&temp->d_lock);
  2079. inode = d_inode(temp);
  2080. if (inode && ceph_snap(inode) == CEPH_SNAPDIR) {
  2081. dout("build_path path+%d: %p SNAPDIR\n",
  2082. pos, temp);
  2083. } else if (stop_on_nosnap && inode && dentry != temp &&
  2084. ceph_snap(inode) == CEPH_NOSNAP) {
  2085. spin_unlock(&temp->d_lock);
  2086. pos++; /* get rid of any prepended '/' */
  2087. break;
  2088. } else {
  2089. pos -= temp->d_name.len;
  2090. if (pos < 0) {
  2091. spin_unlock(&temp->d_lock);
  2092. break;
  2093. }
  2094. memcpy(path + pos, temp->d_name.name, temp->d_name.len);
  2095. }
  2096. spin_unlock(&temp->d_lock);
  2097. temp = READ_ONCE(temp->d_parent);
  2098. /* Are we at the root? */
  2099. if (IS_ROOT(temp))
  2100. break;
  2101. /* Are we out of buffer? */
  2102. if (--pos < 0)
  2103. break;
  2104. path[pos] = '/';
  2105. }
  2106. base = ceph_ino(d_inode(temp));
  2107. rcu_read_unlock();
  2108. if (read_seqretry(&rename_lock, seq))
  2109. goto retry;
  2110. if (pos < 0) {
  2111. /*
  2112. * A rename didn't occur, but somehow we didn't end up where
  2113. * we thought we would. Throw a warning and try again.
  2114. */
  2115. pr_warn("build_path did not end path lookup where "
  2116. "expected, pos is %d\n", pos);
  2117. goto retry;
  2118. }
  2119. *pbase = base;
  2120. *plen = PATH_MAX - 1 - pos;
  2121. dout("build_path on %p %d built %llx '%.*s'\n",
  2122. dentry, d_count(dentry), base, *plen, path + pos);
  2123. return path + pos;
  2124. }
  2125. static int build_dentry_path(struct dentry *dentry, struct inode *dir,
  2126. const char **ppath, int *ppathlen, u64 *pino,
  2127. bool *pfreepath, bool parent_locked)
  2128. {
  2129. char *path;
  2130. rcu_read_lock();
  2131. if (!dir)
  2132. dir = d_inode_rcu(dentry->d_parent);
  2133. if (dir && parent_locked && ceph_snap(dir) == CEPH_NOSNAP) {
  2134. *pino = ceph_ino(dir);
  2135. rcu_read_unlock();
  2136. *ppath = dentry->d_name.name;
  2137. *ppathlen = dentry->d_name.len;
  2138. return 0;
  2139. }
  2140. rcu_read_unlock();
  2141. path = ceph_mdsc_build_path(dentry, ppathlen, pino, 1);
  2142. if (IS_ERR(path))
  2143. return PTR_ERR(path);
  2144. *ppath = path;
  2145. *pfreepath = true;
  2146. return 0;
  2147. }
  2148. static int build_inode_path(struct inode *inode,
  2149. const char **ppath, int *ppathlen, u64 *pino,
  2150. bool *pfreepath)
  2151. {
  2152. struct dentry *dentry;
  2153. char *path;
  2154. if (ceph_snap(inode) == CEPH_NOSNAP) {
  2155. *pino = ceph_ino(inode);
  2156. *ppathlen = 0;
  2157. return 0;
  2158. }
  2159. dentry = d_find_alias(inode);
  2160. path = ceph_mdsc_build_path(dentry, ppathlen, pino, 1);
  2161. dput(dentry);
  2162. if (IS_ERR(path))
  2163. return PTR_ERR(path);
  2164. *ppath = path;
  2165. *pfreepath = true;
  2166. return 0;
  2167. }
  2168. /*
  2169. * request arguments may be specified via an inode *, a dentry *, or
  2170. * an explicit ino+path.
  2171. */
  2172. static int set_request_path_attr(struct inode *rinode, struct dentry *rdentry,
  2173. struct inode *rdiri, const char *rpath,
  2174. u64 rino, const char **ppath, int *pathlen,
  2175. u64 *ino, bool *freepath, bool parent_locked)
  2176. {
  2177. int r = 0;
  2178. if (rinode) {
  2179. r = build_inode_path(rinode, ppath, pathlen, ino, freepath);
  2180. dout(" inode %p %llx.%llx\n", rinode, ceph_ino(rinode),
  2181. ceph_snap(rinode));
  2182. } else if (rdentry) {
  2183. r = build_dentry_path(rdentry, rdiri, ppath, pathlen, ino,
  2184. freepath, parent_locked);
  2185. dout(" dentry %p %llx/%.*s\n", rdentry, *ino, *pathlen,
  2186. *ppath);
  2187. } else if (rpath || rino) {
  2188. *ino = rino;
  2189. *ppath = rpath;
  2190. *pathlen = rpath ? strlen(rpath) : 0;
  2191. dout(" path %.*s\n", *pathlen, rpath);
  2192. }
  2193. return r;
  2194. }
  2195. /*
  2196. * called under mdsc->mutex
  2197. */
  2198. static struct ceph_msg *create_request_message(struct ceph_mds_client *mdsc,
  2199. struct ceph_mds_request *req,
  2200. int mds, bool drop_cap_releases)
  2201. {
  2202. struct ceph_msg *msg;
  2203. struct ceph_mds_request_head *head;
  2204. const char *path1 = NULL;
  2205. const char *path2 = NULL;
  2206. u64 ino1 = 0, ino2 = 0;
  2207. int pathlen1 = 0, pathlen2 = 0;
  2208. bool freepath1 = false, freepath2 = false;
  2209. int len;
  2210. u16 releases;
  2211. void *p, *end;
  2212. int ret;
  2213. ret = set_request_path_attr(req->r_inode, req->r_dentry,
  2214. req->r_parent, req->r_path1, req->r_ino1.ino,
  2215. &path1, &pathlen1, &ino1, &freepath1,
  2216. test_bit(CEPH_MDS_R_PARENT_LOCKED,
  2217. &req->r_req_flags));
  2218. if (ret < 0) {
  2219. msg = ERR_PTR(ret);
  2220. goto out;
  2221. }
  2222. /* If r_old_dentry is set, then assume that its parent is locked */
  2223. ret = set_request_path_attr(NULL, req->r_old_dentry,
  2224. req->r_old_dentry_dir,
  2225. req->r_path2, req->r_ino2.ino,
  2226. &path2, &pathlen2, &ino2, &freepath2, true);
  2227. if (ret < 0) {
  2228. msg = ERR_PTR(ret);
  2229. goto out_free1;
  2230. }
  2231. len = sizeof(*head) +
  2232. pathlen1 + pathlen2 + 2*(1 + sizeof(u32) + sizeof(u64)) +
  2233. sizeof(struct ceph_timespec);
  2234. /* calculate (max) length for cap releases */
  2235. len += sizeof(struct ceph_mds_request_release) *
  2236. (!!req->r_inode_drop + !!req->r_dentry_drop +
  2237. !!req->r_old_inode_drop + !!req->r_old_dentry_drop);
  2238. if (req->r_dentry_drop)
  2239. len += pathlen1;
  2240. if (req->r_old_dentry_drop)
  2241. len += pathlen2;
  2242. msg = ceph_msg_new2(CEPH_MSG_CLIENT_REQUEST, len, 1, GFP_NOFS, false);
  2243. if (!msg) {
  2244. msg = ERR_PTR(-ENOMEM);
  2245. goto out_free2;
  2246. }
  2247. msg->hdr.version = cpu_to_le16(2);
  2248. msg->hdr.tid = cpu_to_le64(req->r_tid);
  2249. head = msg->front.iov_base;
  2250. p = msg->front.iov_base + sizeof(*head);
  2251. end = msg->front.iov_base + msg->front.iov_len;
  2252. head->mdsmap_epoch = cpu_to_le32(mdsc->mdsmap->m_epoch);
  2253. head->op = cpu_to_le32(req->r_op);
  2254. head->caller_uid = cpu_to_le32(from_kuid(&init_user_ns, req->r_uid));
  2255. head->caller_gid = cpu_to_le32(from_kgid(&init_user_ns, req->r_gid));
  2256. head->ino = cpu_to_le64(req->r_deleg_ino);
  2257. head->args = req->r_args;
  2258. ceph_encode_filepath(&p, end, ino1, path1);
  2259. ceph_encode_filepath(&p, end, ino2, path2);
  2260. /* make note of release offset, in case we need to replay */
  2261. req->r_request_release_offset = p - msg->front.iov_base;
  2262. /* cap releases */
  2263. releases = 0;
  2264. if (req->r_inode_drop)
  2265. releases += ceph_encode_inode_release(&p,
  2266. req->r_inode ? req->r_inode : d_inode(req->r_dentry),
  2267. mds, req->r_inode_drop, req->r_inode_unless,
  2268. req->r_op == CEPH_MDS_OP_READDIR);
  2269. if (req->r_dentry_drop)
  2270. releases += ceph_encode_dentry_release(&p, req->r_dentry,
  2271. req->r_parent, mds, req->r_dentry_drop,
  2272. req->r_dentry_unless);
  2273. if (req->r_old_dentry_drop)
  2274. releases += ceph_encode_dentry_release(&p, req->r_old_dentry,
  2275. req->r_old_dentry_dir, mds,
  2276. req->r_old_dentry_drop,
  2277. req->r_old_dentry_unless);
  2278. if (req->r_old_inode_drop)
  2279. releases += ceph_encode_inode_release(&p,
  2280. d_inode(req->r_old_dentry),
  2281. mds, req->r_old_inode_drop, req->r_old_inode_unless, 0);
  2282. if (drop_cap_releases) {
  2283. releases = 0;
  2284. p = msg->front.iov_base + req->r_request_release_offset;
  2285. }
  2286. head->num_releases = cpu_to_le16(releases);
  2287. /* time stamp */
  2288. {
  2289. struct ceph_timespec ts;
  2290. ceph_encode_timespec64(&ts, &req->r_stamp);
  2291. ceph_encode_copy(&p, &ts, sizeof(ts));
  2292. }
  2293. if (WARN_ON_ONCE(p > end)) {
  2294. ceph_msg_put(msg);
  2295. msg = ERR_PTR(-ERANGE);
  2296. goto out_free2;
  2297. }
  2298. msg->front.iov_len = p - msg->front.iov_base;
  2299. msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
  2300. if (req->r_pagelist) {
  2301. struct ceph_pagelist *pagelist = req->r_pagelist;
  2302. ceph_msg_data_add_pagelist(msg, pagelist);
  2303. msg->hdr.data_len = cpu_to_le32(pagelist->length);
  2304. } else {
  2305. msg->hdr.data_len = 0;
  2306. }
  2307. msg->hdr.data_off = cpu_to_le16(0);
  2308. out_free2:
  2309. if (freepath2)
  2310. ceph_mdsc_free_path((char *)path2, pathlen2);
  2311. out_free1:
  2312. if (freepath1)
  2313. ceph_mdsc_free_path((char *)path1, pathlen1);
  2314. out:
  2315. return msg;
  2316. }
  2317. /*
  2318. * called under mdsc->mutex if error, under no mutex if
  2319. * success.
  2320. */
  2321. static void complete_request(struct ceph_mds_client *mdsc,
  2322. struct ceph_mds_request *req)
  2323. {
  2324. req->r_end_latency = ktime_get();
  2325. if (req->r_callback)
  2326. req->r_callback(mdsc, req);
  2327. complete_all(&req->r_completion);
  2328. }
  2329. /*
  2330. * called under mdsc->mutex
  2331. */
  2332. static int __prepare_send_request(struct ceph_mds_client *mdsc,
  2333. struct ceph_mds_request *req,
  2334. int mds, bool drop_cap_releases)
  2335. {
  2336. struct ceph_mds_request_head *rhead;
  2337. struct ceph_msg *msg;
  2338. int flags = 0;
  2339. req->r_attempts++;
  2340. if (req->r_inode) {
  2341. struct ceph_cap *cap =
  2342. ceph_get_cap_for_mds(ceph_inode(req->r_inode), mds);
  2343. if (cap)
  2344. req->r_sent_on_mseq = cap->mseq;
  2345. else
  2346. req->r_sent_on_mseq = -1;
  2347. }
  2348. dout("prepare_send_request %p tid %lld %s (attempt %d)\n", req,
  2349. req->r_tid, ceph_mds_op_name(req->r_op), req->r_attempts);
  2350. if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
  2351. void *p;
  2352. /*
  2353. * Replay. Do not regenerate message (and rebuild
  2354. * paths, etc.); just use the original message.
  2355. * Rebuilding paths will break for renames because
  2356. * d_move mangles the src name.
  2357. */
  2358. msg = req->r_request;
  2359. rhead = msg->front.iov_base;
  2360. flags = le32_to_cpu(rhead->flags);
  2361. flags |= CEPH_MDS_FLAG_REPLAY;
  2362. rhead->flags = cpu_to_le32(flags);
  2363. if (req->r_target_inode)
  2364. rhead->ino = cpu_to_le64(ceph_ino(req->r_target_inode));
  2365. rhead->num_retry = req->r_attempts - 1;
  2366. /* remove cap/dentry releases from message */
  2367. rhead->num_releases = 0;
  2368. /* time stamp */
  2369. p = msg->front.iov_base + req->r_request_release_offset;
  2370. {
  2371. struct ceph_timespec ts;
  2372. ceph_encode_timespec64(&ts, &req->r_stamp);
  2373. ceph_encode_copy(&p, &ts, sizeof(ts));
  2374. }
  2375. msg->front.iov_len = p - msg->front.iov_base;
  2376. msg->hdr.front_len = cpu_to_le32(msg->front.iov_len);
  2377. return 0;
  2378. }
  2379. if (req->r_request) {
  2380. ceph_msg_put(req->r_request);
  2381. req->r_request = NULL;
  2382. }
  2383. msg = create_request_message(mdsc, req, mds, drop_cap_releases);
  2384. if (IS_ERR(msg)) {
  2385. req->r_err = PTR_ERR(msg);
  2386. return PTR_ERR(msg);
  2387. }
  2388. req->r_request = msg;
  2389. rhead = msg->front.iov_base;
  2390. rhead->oldest_client_tid = cpu_to_le64(__get_oldest_tid(mdsc));
  2391. if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
  2392. flags |= CEPH_MDS_FLAG_REPLAY;
  2393. if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags))
  2394. flags |= CEPH_MDS_FLAG_ASYNC;
  2395. if (req->r_parent)
  2396. flags |= CEPH_MDS_FLAG_WANT_DENTRY;
  2397. rhead->flags = cpu_to_le32(flags);
  2398. rhead->num_fwd = req->r_num_fwd;
  2399. rhead->num_retry = req->r_attempts - 1;
  2400. dout(" r_parent = %p\n", req->r_parent);
  2401. return 0;
  2402. }
  2403. /*
  2404. * called under mdsc->mutex
  2405. */
  2406. static int __send_request(struct ceph_mds_client *mdsc,
  2407. struct ceph_mds_session *session,
  2408. struct ceph_mds_request *req,
  2409. bool drop_cap_releases)
  2410. {
  2411. int err;
  2412. err = __prepare_send_request(mdsc, req, session->s_mds,
  2413. drop_cap_releases);
  2414. if (!err) {
  2415. ceph_msg_get(req->r_request);
  2416. ceph_con_send(&session->s_con, req->r_request);
  2417. }
  2418. return err;
  2419. }
  2420. /*
  2421. * send request, or put it on the appropriate wait list.
  2422. */
  2423. static void __do_request(struct ceph_mds_client *mdsc,
  2424. struct ceph_mds_request *req)
  2425. {
  2426. struct ceph_mds_session *session = NULL;
  2427. int mds = -1;
  2428. int err = 0;
  2429. bool random;
  2430. if (req->r_err || test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) {
  2431. if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags))
  2432. __unregister_request(mdsc, req);
  2433. return;
  2434. }
  2435. if (req->r_timeout &&
  2436. time_after_eq(jiffies, req->r_started + req->r_timeout)) {
  2437. dout("do_request timed out\n");
  2438. err = -ETIMEDOUT;
  2439. goto finish;
  2440. }
  2441. if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN) {
  2442. dout("do_request forced umount\n");
  2443. err = -EIO;
  2444. goto finish;
  2445. }
  2446. if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_MOUNTING) {
  2447. if (mdsc->mdsmap_err) {
  2448. err = mdsc->mdsmap_err;
  2449. dout("do_request mdsmap err %d\n", err);
  2450. goto finish;
  2451. }
  2452. if (mdsc->mdsmap->m_epoch == 0) {
  2453. dout("do_request no mdsmap, waiting for map\n");
  2454. list_add(&req->r_wait, &mdsc->waiting_for_map);
  2455. return;
  2456. }
  2457. if (!(mdsc->fsc->mount_options->flags &
  2458. CEPH_MOUNT_OPT_MOUNTWAIT) &&
  2459. !ceph_mdsmap_is_cluster_available(mdsc->mdsmap)) {
  2460. err = -EHOSTUNREACH;
  2461. goto finish;
  2462. }
  2463. }
  2464. put_request_session(req);
  2465. mds = __choose_mds(mdsc, req, &random);
  2466. if (mds < 0 ||
  2467. ceph_mdsmap_get_state(mdsc->mdsmap, mds) < CEPH_MDS_STATE_ACTIVE) {
  2468. if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags)) {
  2469. err = -EJUKEBOX;
  2470. goto finish;
  2471. }
  2472. dout("do_request no mds or not active, waiting for map\n");
  2473. list_add(&req->r_wait, &mdsc->waiting_for_map);
  2474. return;
  2475. }
  2476. /* get, open session */
  2477. session = __ceph_lookup_mds_session(mdsc, mds);
  2478. if (!session) {
  2479. session = register_session(mdsc, mds);
  2480. if (IS_ERR(session)) {
  2481. err = PTR_ERR(session);
  2482. goto finish;
  2483. }
  2484. }
  2485. req->r_session = ceph_get_mds_session(session);
  2486. dout("do_request mds%d session %p state %s\n", mds, session,
  2487. ceph_session_state_name(session->s_state));
  2488. if (session->s_state != CEPH_MDS_SESSION_OPEN &&
  2489. session->s_state != CEPH_MDS_SESSION_HUNG) {
  2490. if (session->s_state == CEPH_MDS_SESSION_REJECTED) {
  2491. err = -EACCES;
  2492. goto out_session;
  2493. }
  2494. /*
  2495. * We cannot queue async requests since the caps and delegated
  2496. * inodes are bound to the session. Just return -EJUKEBOX and
  2497. * let the caller retry a sync request in that case.
  2498. */
  2499. if (test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags)) {
  2500. err = -EJUKEBOX;
  2501. goto out_session;
  2502. }
  2503. if (session->s_state == CEPH_MDS_SESSION_NEW ||
  2504. session->s_state == CEPH_MDS_SESSION_CLOSING) {
  2505. err = __open_session(mdsc, session);
  2506. if (err)
  2507. goto out_session;
  2508. /* retry the same mds later */
  2509. if (random)
  2510. req->r_resend_mds = mds;
  2511. }
  2512. list_add(&req->r_wait, &session->s_waiting);
  2513. goto out_session;
  2514. }
  2515. /* send request */
  2516. req->r_resend_mds = -1; /* forget any previous mds hint */
  2517. if (req->r_request_started == 0) /* note request start time */
  2518. req->r_request_started = jiffies;
  2519. err = __send_request(mdsc, session, req, false);
  2520. out_session:
  2521. ceph_put_mds_session(session);
  2522. finish:
  2523. if (err) {
  2524. dout("__do_request early error %d\n", err);
  2525. req->r_err = err;
  2526. complete_request(mdsc, req);
  2527. __unregister_request(mdsc, req);
  2528. }
  2529. return;
  2530. }
  2531. /*
  2532. * called under mdsc->mutex
  2533. */
  2534. static void __wake_requests(struct ceph_mds_client *mdsc,
  2535. struct list_head *head)
  2536. {
  2537. struct ceph_mds_request *req;
  2538. LIST_HEAD(tmp_list);
  2539. list_splice_init(head, &tmp_list);
  2540. while (!list_empty(&tmp_list)) {
  2541. req = list_entry(tmp_list.next,
  2542. struct ceph_mds_request, r_wait);
  2543. list_del_init(&req->r_wait);
  2544. dout(" wake request %p tid %llu\n", req, req->r_tid);
  2545. __do_request(mdsc, req);
  2546. }
  2547. }
  2548. /*
  2549. * Wake up threads with requests pending for @mds, so that they can
  2550. * resubmit their requests to a possibly different mds.
  2551. */
  2552. static void kick_requests(struct ceph_mds_client *mdsc, int mds)
  2553. {
  2554. struct ceph_mds_request *req;
  2555. struct rb_node *p = rb_first(&mdsc->request_tree);
  2556. dout("kick_requests mds%d\n", mds);
  2557. while (p) {
  2558. req = rb_entry(p, struct ceph_mds_request, r_node);
  2559. p = rb_next(p);
  2560. if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
  2561. continue;
  2562. if (req->r_attempts > 0)
  2563. continue; /* only new requests */
  2564. if (req->r_session &&
  2565. req->r_session->s_mds == mds) {
  2566. dout(" kicking tid %llu\n", req->r_tid);
  2567. list_del_init(&req->r_wait);
  2568. __do_request(mdsc, req);
  2569. }
  2570. }
  2571. }
  2572. int ceph_mdsc_submit_request(struct ceph_mds_client *mdsc, struct inode *dir,
  2573. struct ceph_mds_request *req)
  2574. {
  2575. int err = 0;
  2576. /* take CAP_PIN refs for r_inode, r_parent, r_old_dentry */
  2577. if (req->r_inode)
  2578. ceph_get_cap_refs(ceph_inode(req->r_inode), CEPH_CAP_PIN);
  2579. if (req->r_parent) {
  2580. struct ceph_inode_info *ci = ceph_inode(req->r_parent);
  2581. int fmode = (req->r_op & CEPH_MDS_OP_WRITE) ?
  2582. CEPH_FILE_MODE_WR : CEPH_FILE_MODE_RD;
  2583. spin_lock(&ci->i_ceph_lock);
  2584. ceph_take_cap_refs(ci, CEPH_CAP_PIN, false);
  2585. __ceph_touch_fmode(ci, mdsc, fmode);
  2586. spin_unlock(&ci->i_ceph_lock);
  2587. ihold(req->r_parent);
  2588. }
  2589. if (req->r_old_dentry_dir)
  2590. ceph_get_cap_refs(ceph_inode(req->r_old_dentry_dir),
  2591. CEPH_CAP_PIN);
  2592. if (req->r_inode) {
  2593. err = ceph_wait_on_async_create(req->r_inode);
  2594. if (err) {
  2595. dout("%s: wait for async create returned: %d\n",
  2596. __func__, err);
  2597. return err;
  2598. }
  2599. }
  2600. if (!err && req->r_old_inode) {
  2601. err = ceph_wait_on_async_create(req->r_old_inode);
  2602. if (err) {
  2603. dout("%s: wait for async create returned: %d\n",
  2604. __func__, err);
  2605. return err;
  2606. }
  2607. }
  2608. dout("submit_request on %p for inode %p\n", req, dir);
  2609. mutex_lock(&mdsc->mutex);
  2610. __register_request(mdsc, req, dir);
  2611. __do_request(mdsc, req);
  2612. err = req->r_err;
  2613. mutex_unlock(&mdsc->mutex);
  2614. return err;
  2615. }
  2616. static int ceph_mdsc_wait_request(struct ceph_mds_client *mdsc,
  2617. struct ceph_mds_request *req)
  2618. {
  2619. int err;
  2620. /* wait */
  2621. dout("do_request waiting\n");
  2622. if (!req->r_timeout && req->r_wait_for_completion) {
  2623. err = req->r_wait_for_completion(mdsc, req);
  2624. } else {
  2625. long timeleft = wait_for_completion_killable_timeout(
  2626. &req->r_completion,
  2627. ceph_timeout_jiffies(req->r_timeout));
  2628. if (timeleft > 0)
  2629. err = 0;
  2630. else if (!timeleft)
  2631. err = -ETIMEDOUT; /* timed out */
  2632. else
  2633. err = timeleft; /* killed */
  2634. }
  2635. dout("do_request waited, got %d\n", err);
  2636. mutex_lock(&mdsc->mutex);
  2637. /* only abort if we didn't race with a real reply */
  2638. if (test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags)) {
  2639. err = le32_to_cpu(req->r_reply_info.head->result);
  2640. } else if (err < 0) {
  2641. dout("aborted request %lld with %d\n", req->r_tid, err);
  2642. /*
  2643. * ensure we aren't running concurrently with
  2644. * ceph_fill_trace or ceph_readdir_prepopulate, which
  2645. * rely on locks (dir mutex) held by our caller.
  2646. */
  2647. mutex_lock(&req->r_fill_mutex);
  2648. req->r_err = err;
  2649. set_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags);
  2650. mutex_unlock(&req->r_fill_mutex);
  2651. if (req->r_parent &&
  2652. (req->r_op & CEPH_MDS_OP_WRITE))
  2653. ceph_invalidate_dir_request(req);
  2654. } else {
  2655. err = req->r_err;
  2656. }
  2657. mutex_unlock(&mdsc->mutex);
  2658. return err;
  2659. }
  2660. /*
  2661. * Synchrously perform an mds request. Take care of all of the
  2662. * session setup, forwarding, retry details.
  2663. */
  2664. int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
  2665. struct inode *dir,
  2666. struct ceph_mds_request *req)
  2667. {
  2668. int err;
  2669. dout("do_request on %p\n", req);
  2670. /* issue */
  2671. err = ceph_mdsc_submit_request(mdsc, dir, req);
  2672. if (!err)
  2673. err = ceph_mdsc_wait_request(mdsc, req);
  2674. dout("do_request %p done, result %d\n", req, err);
  2675. return err;
  2676. }
  2677. /*
  2678. * Invalidate dir's completeness, dentry lease state on an aborted MDS
  2679. * namespace request.
  2680. */
  2681. void ceph_invalidate_dir_request(struct ceph_mds_request *req)
  2682. {
  2683. struct inode *dir = req->r_parent;
  2684. struct inode *old_dir = req->r_old_dentry_dir;
  2685. dout("invalidate_dir_request %p %p (complete, lease(s))\n", dir, old_dir);
  2686. ceph_dir_clear_complete(dir);
  2687. if (old_dir)
  2688. ceph_dir_clear_complete(old_dir);
  2689. if (req->r_dentry)
  2690. ceph_invalidate_dentry_lease(req->r_dentry);
  2691. if (req->r_old_dentry)
  2692. ceph_invalidate_dentry_lease(req->r_old_dentry);
  2693. }
  2694. /*
  2695. * Handle mds reply.
  2696. *
  2697. * We take the session mutex and parse and process the reply immediately.
  2698. * This preserves the logical ordering of replies, capabilities, etc., sent
  2699. * by the MDS as they are applied to our local cache.
  2700. */
  2701. static void handle_reply(struct ceph_mds_session *session, struct ceph_msg *msg)
  2702. {
  2703. struct ceph_mds_client *mdsc = session->s_mdsc;
  2704. struct ceph_mds_request *req;
  2705. struct ceph_mds_reply_head *head = msg->front.iov_base;
  2706. struct ceph_mds_reply_info_parsed *rinfo; /* parsed reply info */
  2707. struct ceph_snap_realm *realm;
  2708. u64 tid;
  2709. int err, result;
  2710. int mds = session->s_mds;
  2711. if (msg->front.iov_len < sizeof(*head)) {
  2712. pr_err("mdsc_handle_reply got corrupt (short) reply\n");
  2713. ceph_msg_dump(msg);
  2714. return;
  2715. }
  2716. /* get request, session */
  2717. tid = le64_to_cpu(msg->hdr.tid);
  2718. mutex_lock(&mdsc->mutex);
  2719. req = lookup_get_request(mdsc, tid);
  2720. if (!req) {
  2721. dout("handle_reply on unknown tid %llu\n", tid);
  2722. mutex_unlock(&mdsc->mutex);
  2723. return;
  2724. }
  2725. dout("handle_reply %p\n", req);
  2726. /* correct session? */
  2727. if (req->r_session != session) {
  2728. pr_err("mdsc_handle_reply got %llu on session mds%d"
  2729. " not mds%d\n", tid, session->s_mds,
  2730. req->r_session ? req->r_session->s_mds : -1);
  2731. mutex_unlock(&mdsc->mutex);
  2732. goto out;
  2733. }
  2734. /* dup? */
  2735. if ((test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags) && !head->safe) ||
  2736. (test_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags) && head->safe)) {
  2737. pr_warn("got a dup %s reply on %llu from mds%d\n",
  2738. head->safe ? "safe" : "unsafe", tid, mds);
  2739. mutex_unlock(&mdsc->mutex);
  2740. goto out;
  2741. }
  2742. if (test_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags)) {
  2743. pr_warn("got unsafe after safe on %llu from mds%d\n",
  2744. tid, mds);
  2745. mutex_unlock(&mdsc->mutex);
  2746. goto out;
  2747. }
  2748. result = le32_to_cpu(head->result);
  2749. /*
  2750. * Handle an ESTALE
  2751. * if we're not talking to the authority, send to them
  2752. * if the authority has changed while we weren't looking,
  2753. * send to new authority
  2754. * Otherwise we just have to return an ESTALE
  2755. */
  2756. if (result == -ESTALE) {
  2757. dout("got ESTALE on request %llu\n", req->r_tid);
  2758. req->r_resend_mds = -1;
  2759. if (req->r_direct_mode != USE_AUTH_MDS) {
  2760. dout("not using auth, setting for that now\n");
  2761. req->r_direct_mode = USE_AUTH_MDS;
  2762. __do_request(mdsc, req);
  2763. mutex_unlock(&mdsc->mutex);
  2764. goto out;
  2765. } else {
  2766. int mds = __choose_mds(mdsc, req, NULL);
  2767. if (mds >= 0 && mds != req->r_session->s_mds) {
  2768. dout("but auth changed, so resending\n");
  2769. __do_request(mdsc, req);
  2770. mutex_unlock(&mdsc->mutex);
  2771. goto out;
  2772. }
  2773. }
  2774. dout("have to return ESTALE on request %llu\n", req->r_tid);
  2775. }
  2776. if (head->safe) {
  2777. set_bit(CEPH_MDS_R_GOT_SAFE, &req->r_req_flags);
  2778. __unregister_request(mdsc, req);
  2779. /* last request during umount? */
  2780. if (mdsc->stopping && !__get_oldest_req(mdsc))
  2781. complete_all(&mdsc->safe_umount_waiters);
  2782. if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
  2783. /*
  2784. * We already handled the unsafe response, now do the
  2785. * cleanup. No need to examine the response; the MDS
  2786. * doesn't include any result info in the safe
  2787. * response. And even if it did, there is nothing
  2788. * useful we could do with a revised return value.
  2789. */
  2790. dout("got safe reply %llu, mds%d\n", tid, mds);
  2791. mutex_unlock(&mdsc->mutex);
  2792. goto out;
  2793. }
  2794. } else {
  2795. set_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags);
  2796. list_add_tail(&req->r_unsafe_item, &req->r_session->s_unsafe);
  2797. }
  2798. dout("handle_reply tid %lld result %d\n", tid, result);
  2799. rinfo = &req->r_reply_info;
  2800. if (test_bit(CEPHFS_FEATURE_REPLY_ENCODING, &session->s_features))
  2801. err = parse_reply_info(session, msg, rinfo, (u64)-1);
  2802. else
  2803. err = parse_reply_info(session, msg, rinfo, session->s_con.peer_features);
  2804. mutex_unlock(&mdsc->mutex);
  2805. mutex_lock(&session->s_mutex);
  2806. if (err < 0) {
  2807. pr_err("mdsc_handle_reply got corrupt reply mds%d(tid:%lld)\n", mds, tid);
  2808. ceph_msg_dump(msg);
  2809. goto out_err;
  2810. }
  2811. /* snap trace */
  2812. realm = NULL;
  2813. if (rinfo->snapblob_len) {
  2814. down_write(&mdsc->snap_rwsem);
  2815. ceph_update_snap_trace(mdsc, rinfo->snapblob,
  2816. rinfo->snapblob + rinfo->snapblob_len,
  2817. le32_to_cpu(head->op) == CEPH_MDS_OP_RMSNAP,
  2818. &realm);
  2819. downgrade_write(&mdsc->snap_rwsem);
  2820. } else {
  2821. down_read(&mdsc->snap_rwsem);
  2822. }
  2823. /* insert trace into our cache */
  2824. mutex_lock(&req->r_fill_mutex);
  2825. current->journal_info = req;
  2826. err = ceph_fill_trace(mdsc->fsc->sb, req);
  2827. if (err == 0) {
  2828. if (result == 0 && (req->r_op == CEPH_MDS_OP_READDIR ||
  2829. req->r_op == CEPH_MDS_OP_LSSNAP))
  2830. ceph_readdir_prepopulate(req, req->r_session);
  2831. }
  2832. current->journal_info = NULL;
  2833. mutex_unlock(&req->r_fill_mutex);
  2834. up_read(&mdsc->snap_rwsem);
  2835. if (realm)
  2836. ceph_put_snap_realm(mdsc, realm);
  2837. if (err == 0) {
  2838. if (req->r_target_inode &&
  2839. test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags)) {
  2840. struct ceph_inode_info *ci =
  2841. ceph_inode(req->r_target_inode);
  2842. spin_lock(&ci->i_unsafe_lock);
  2843. list_add_tail(&req->r_unsafe_target_item,
  2844. &ci->i_unsafe_iops);
  2845. spin_unlock(&ci->i_unsafe_lock);
  2846. }
  2847. ceph_unreserve_caps(mdsc, &req->r_caps_reservation);
  2848. }
  2849. out_err:
  2850. mutex_lock(&mdsc->mutex);
  2851. if (!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
  2852. if (err) {
  2853. req->r_err = err;
  2854. } else {
  2855. req->r_reply = ceph_msg_get(msg);
  2856. set_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags);
  2857. }
  2858. } else {
  2859. dout("reply arrived after request %lld was aborted\n", tid);
  2860. }
  2861. mutex_unlock(&mdsc->mutex);
  2862. mutex_unlock(&session->s_mutex);
  2863. /* kick calling process */
  2864. complete_request(mdsc, req);
  2865. ceph_update_metadata_latency(&mdsc->metric, req->r_start_latency,
  2866. req->r_end_latency, err);
  2867. out:
  2868. ceph_mdsc_put_request(req);
  2869. return;
  2870. }
  2871. /*
  2872. * handle mds notification that our request has been forwarded.
  2873. */
  2874. static void handle_forward(struct ceph_mds_client *mdsc,
  2875. struct ceph_mds_session *session,
  2876. struct ceph_msg *msg)
  2877. {
  2878. struct ceph_mds_request *req;
  2879. u64 tid = le64_to_cpu(msg->hdr.tid);
  2880. u32 next_mds;
  2881. u32 fwd_seq;
  2882. int err = -EINVAL;
  2883. void *p = msg->front.iov_base;
  2884. void *end = p + msg->front.iov_len;
  2885. ceph_decode_need(&p, end, 2*sizeof(u32), bad);
  2886. next_mds = ceph_decode_32(&p);
  2887. fwd_seq = ceph_decode_32(&p);
  2888. mutex_lock(&mdsc->mutex);
  2889. req = lookup_get_request(mdsc, tid);
  2890. if (!req) {
  2891. dout("forward tid %llu to mds%d - req dne\n", tid, next_mds);
  2892. goto out; /* dup reply? */
  2893. }
  2894. if (test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags)) {
  2895. dout("forward tid %llu aborted, unregistering\n", tid);
  2896. __unregister_request(mdsc, req);
  2897. } else if (fwd_seq <= req->r_num_fwd) {
  2898. dout("forward tid %llu to mds%d - old seq %d <= %d\n",
  2899. tid, next_mds, req->r_num_fwd, fwd_seq);
  2900. } else {
  2901. /* resend. forward race not possible; mds would drop */
  2902. dout("forward tid %llu to mds%d (we resend)\n", tid, next_mds);
  2903. BUG_ON(req->r_err);
  2904. BUG_ON(test_bit(CEPH_MDS_R_GOT_RESULT, &req->r_req_flags));
  2905. req->r_attempts = 0;
  2906. req->r_num_fwd = fwd_seq;
  2907. req->r_resend_mds = next_mds;
  2908. put_request_session(req);
  2909. __do_request(mdsc, req);
  2910. }
  2911. ceph_mdsc_put_request(req);
  2912. out:
  2913. mutex_unlock(&mdsc->mutex);
  2914. return;
  2915. bad:
  2916. pr_err("mdsc_handle_forward decode error err=%d\n", err);
  2917. }
  2918. static int __decode_session_metadata(void **p, void *end,
  2919. bool *blocklisted)
  2920. {
  2921. /* map<string,string> */
  2922. u32 n;
  2923. bool err_str;
  2924. ceph_decode_32_safe(p, end, n, bad);
  2925. while (n-- > 0) {
  2926. u32 len;
  2927. ceph_decode_32_safe(p, end, len, bad);
  2928. ceph_decode_need(p, end, len, bad);
  2929. err_str = !strncmp(*p, "error_string", len);
  2930. *p += len;
  2931. ceph_decode_32_safe(p, end, len, bad);
  2932. ceph_decode_need(p, end, len, bad);
  2933. /*
  2934. * Match "blocklisted (blacklisted)" from newer MDSes,
  2935. * or "blacklisted" from older MDSes.
  2936. */
  2937. if (err_str && strnstr(*p, "blacklisted", len))
  2938. *blocklisted = true;
  2939. *p += len;
  2940. }
  2941. return 0;
  2942. bad:
  2943. return -1;
  2944. }
  2945. /*
  2946. * handle a mds session control message
  2947. */
  2948. static void handle_session(struct ceph_mds_session *session,
  2949. struct ceph_msg *msg)
  2950. {
  2951. struct ceph_mds_client *mdsc = session->s_mdsc;
  2952. int mds = session->s_mds;
  2953. int msg_version = le16_to_cpu(msg->hdr.version);
  2954. void *p = msg->front.iov_base;
  2955. void *end = p + msg->front.iov_len;
  2956. struct ceph_mds_session_head *h;
  2957. u32 op;
  2958. u64 seq, features = 0;
  2959. int wake = 0;
  2960. bool blocklisted = false;
  2961. /* decode */
  2962. ceph_decode_need(&p, end, sizeof(*h), bad);
  2963. h = p;
  2964. p += sizeof(*h);
  2965. op = le32_to_cpu(h->op);
  2966. seq = le64_to_cpu(h->seq);
  2967. if (msg_version >= 3) {
  2968. u32 len;
  2969. /* version >= 2, metadata */
  2970. if (__decode_session_metadata(&p, end, &blocklisted) < 0)
  2971. goto bad;
  2972. /* version >= 3, feature bits */
  2973. ceph_decode_32_safe(&p, end, len, bad);
  2974. if (len) {
  2975. ceph_decode_64_safe(&p, end, features, bad);
  2976. p += len - sizeof(features);
  2977. }
  2978. }
  2979. mutex_lock(&mdsc->mutex);
  2980. if (op == CEPH_SESSION_CLOSE) {
  2981. ceph_get_mds_session(session);
  2982. __unregister_session(mdsc, session);
  2983. }
  2984. /* FIXME: this ttl calculation is generous */
  2985. session->s_ttl = jiffies + HZ*mdsc->mdsmap->m_session_autoclose;
  2986. mutex_unlock(&mdsc->mutex);
  2987. mutex_lock(&session->s_mutex);
  2988. dout("handle_session mds%d %s %p state %s seq %llu\n",
  2989. mds, ceph_session_op_name(op), session,
  2990. ceph_session_state_name(session->s_state), seq);
  2991. if (session->s_state == CEPH_MDS_SESSION_HUNG) {
  2992. session->s_state = CEPH_MDS_SESSION_OPEN;
  2993. pr_info("mds%d came back\n", session->s_mds);
  2994. }
  2995. switch (op) {
  2996. case CEPH_SESSION_OPEN:
  2997. if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
  2998. pr_info("mds%d reconnect success\n", session->s_mds);
  2999. session->s_state = CEPH_MDS_SESSION_OPEN;
  3000. session->s_features = features;
  3001. renewed_caps(mdsc, session, 0);
  3002. if (test_bit(CEPHFS_FEATURE_METRIC_COLLECT, &session->s_features))
  3003. metric_schedule_delayed(&mdsc->metric);
  3004. wake = 1;
  3005. if (mdsc->stopping)
  3006. __close_session(mdsc, session);
  3007. break;
  3008. case CEPH_SESSION_RENEWCAPS:
  3009. if (session->s_renew_seq == seq)
  3010. renewed_caps(mdsc, session, 1);
  3011. break;
  3012. case CEPH_SESSION_CLOSE:
  3013. if (session->s_state == CEPH_MDS_SESSION_RECONNECTING)
  3014. pr_info("mds%d reconnect denied\n", session->s_mds);
  3015. session->s_state = CEPH_MDS_SESSION_CLOSED;
  3016. cleanup_session_requests(mdsc, session);
  3017. remove_session_caps(session);
  3018. wake = 2; /* for good measure */
  3019. wake_up_all(&mdsc->session_close_wq);
  3020. break;
  3021. case CEPH_SESSION_STALE:
  3022. pr_info("mds%d caps went stale, renewing\n",
  3023. session->s_mds);
  3024. spin_lock(&session->s_gen_ttl_lock);
  3025. session->s_cap_gen++;
  3026. session->s_cap_ttl = jiffies - 1;
  3027. spin_unlock(&session->s_gen_ttl_lock);
  3028. send_renew_caps(mdsc, session);
  3029. break;
  3030. case CEPH_SESSION_RECALL_STATE:
  3031. ceph_trim_caps(mdsc, session, le32_to_cpu(h->max_caps));
  3032. break;
  3033. case CEPH_SESSION_FLUSHMSG:
  3034. send_flushmsg_ack(mdsc, session, seq);
  3035. break;
  3036. case CEPH_SESSION_FORCE_RO:
  3037. dout("force_session_readonly %p\n", session);
  3038. spin_lock(&session->s_cap_lock);
  3039. session->s_readonly = true;
  3040. spin_unlock(&session->s_cap_lock);
  3041. wake_up_session_caps(session, FORCE_RO);
  3042. break;
  3043. case CEPH_SESSION_REJECT:
  3044. WARN_ON(session->s_state != CEPH_MDS_SESSION_OPENING);
  3045. pr_info("mds%d rejected session\n", session->s_mds);
  3046. session->s_state = CEPH_MDS_SESSION_REJECTED;
  3047. cleanup_session_requests(mdsc, session);
  3048. remove_session_caps(session);
  3049. if (blocklisted)
  3050. mdsc->fsc->blocklisted = true;
  3051. wake = 2; /* for good measure */
  3052. break;
  3053. default:
  3054. pr_err("mdsc_handle_session bad op %d mds%d\n", op, mds);
  3055. WARN_ON(1);
  3056. }
  3057. mutex_unlock(&session->s_mutex);
  3058. if (wake) {
  3059. mutex_lock(&mdsc->mutex);
  3060. __wake_requests(mdsc, &session->s_waiting);
  3061. if (wake == 2)
  3062. kick_requests(mdsc, mds);
  3063. mutex_unlock(&mdsc->mutex);
  3064. }
  3065. if (op == CEPH_SESSION_CLOSE)
  3066. ceph_put_mds_session(session);
  3067. return;
  3068. bad:
  3069. pr_err("mdsc_handle_session corrupt message mds%d len %d\n", mds,
  3070. (int)msg->front.iov_len);
  3071. ceph_msg_dump(msg);
  3072. return;
  3073. }
  3074. void ceph_mdsc_release_dir_caps(struct ceph_mds_request *req)
  3075. {
  3076. int dcaps;
  3077. dcaps = xchg(&req->r_dir_caps, 0);
  3078. if (dcaps) {
  3079. dout("releasing r_dir_caps=%s\n", ceph_cap_string(dcaps));
  3080. ceph_put_cap_refs(ceph_inode(req->r_parent), dcaps);
  3081. }
  3082. }
  3083. void ceph_mdsc_release_dir_caps_no_check(struct ceph_mds_request *req)
  3084. {
  3085. int dcaps;
  3086. dcaps = xchg(&req->r_dir_caps, 0);
  3087. if (dcaps) {
  3088. dout("releasing r_dir_caps=%s\n", ceph_cap_string(dcaps));
  3089. ceph_put_cap_refs_no_check_caps(ceph_inode(req->r_parent),
  3090. dcaps);
  3091. }
  3092. }
  3093. /*
  3094. * called under session->mutex.
  3095. */
  3096. static void replay_unsafe_requests(struct ceph_mds_client *mdsc,
  3097. struct ceph_mds_session *session)
  3098. {
  3099. struct ceph_mds_request *req, *nreq;
  3100. struct rb_node *p;
  3101. dout("replay_unsafe_requests mds%d\n", session->s_mds);
  3102. mutex_lock(&mdsc->mutex);
  3103. list_for_each_entry_safe(req, nreq, &session->s_unsafe, r_unsafe_item)
  3104. __send_request(mdsc, session, req, true);
  3105. /*
  3106. * also re-send old requests when MDS enters reconnect stage. So that MDS
  3107. * can process completed request in clientreplay stage.
  3108. */
  3109. p = rb_first(&mdsc->request_tree);
  3110. while (p) {
  3111. req = rb_entry(p, struct ceph_mds_request, r_node);
  3112. p = rb_next(p);
  3113. if (test_bit(CEPH_MDS_R_GOT_UNSAFE, &req->r_req_flags))
  3114. continue;
  3115. if (req->r_attempts == 0)
  3116. continue; /* only old requests */
  3117. if (!req->r_session)
  3118. continue;
  3119. if (req->r_session->s_mds != session->s_mds)
  3120. continue;
  3121. ceph_mdsc_release_dir_caps_no_check(req);
  3122. __send_request(mdsc, session, req, true);
  3123. }
  3124. mutex_unlock(&mdsc->mutex);
  3125. }
  3126. static int send_reconnect_partial(struct ceph_reconnect_state *recon_state)
  3127. {
  3128. struct ceph_msg *reply;
  3129. struct ceph_pagelist *_pagelist;
  3130. struct page *page;
  3131. __le32 *addr;
  3132. int err = -ENOMEM;
  3133. if (!recon_state->allow_multi)
  3134. return -ENOSPC;
  3135. /* can't handle message that contains both caps and realm */
  3136. BUG_ON(!recon_state->nr_caps == !recon_state->nr_realms);
  3137. /* pre-allocate new pagelist */
  3138. _pagelist = ceph_pagelist_alloc(GFP_NOFS);
  3139. if (!_pagelist)
  3140. return -ENOMEM;
  3141. reply = ceph_msg_new2(CEPH_MSG_CLIENT_RECONNECT, 0, 1, GFP_NOFS, false);
  3142. if (!reply)
  3143. goto fail_msg;
  3144. /* placeholder for nr_caps */
  3145. err = ceph_pagelist_encode_32(_pagelist, 0);
  3146. if (err < 0)
  3147. goto fail;
  3148. if (recon_state->nr_caps) {
  3149. /* currently encoding caps */
  3150. err = ceph_pagelist_encode_32(recon_state->pagelist, 0);
  3151. if (err)
  3152. goto fail;
  3153. } else {
  3154. /* placeholder for nr_realms (currently encoding relams) */
  3155. err = ceph_pagelist_encode_32(_pagelist, 0);
  3156. if (err < 0)
  3157. goto fail;
  3158. }
  3159. err = ceph_pagelist_encode_8(recon_state->pagelist, 1);
  3160. if (err)
  3161. goto fail;
  3162. page = list_first_entry(&recon_state->pagelist->head, struct page, lru);
  3163. addr = kmap_atomic(page);
  3164. if (recon_state->nr_caps) {
  3165. /* currently encoding caps */
  3166. *addr = cpu_to_le32(recon_state->nr_caps);
  3167. } else {
  3168. /* currently encoding relams */
  3169. *(addr + 1) = cpu_to_le32(recon_state->nr_realms);
  3170. }
  3171. kunmap_atomic(addr);
  3172. reply->hdr.version = cpu_to_le16(5);
  3173. reply->hdr.compat_version = cpu_to_le16(4);
  3174. reply->hdr.data_len = cpu_to_le32(recon_state->pagelist->length);
  3175. ceph_msg_data_add_pagelist(reply, recon_state->pagelist);
  3176. ceph_con_send(&recon_state->session->s_con, reply);
  3177. ceph_pagelist_release(recon_state->pagelist);
  3178. recon_state->pagelist = _pagelist;
  3179. recon_state->nr_caps = 0;
  3180. recon_state->nr_realms = 0;
  3181. recon_state->msg_version = 5;
  3182. return 0;
  3183. fail:
  3184. ceph_msg_put(reply);
  3185. fail_msg:
  3186. ceph_pagelist_release(_pagelist);
  3187. return err;
  3188. }
  3189. static struct dentry* d_find_primary(struct inode *inode)
  3190. {
  3191. struct dentry *alias, *dn = NULL;
  3192. if (hlist_empty(&inode->i_dentry))
  3193. return NULL;
  3194. spin_lock(&inode->i_lock);
  3195. if (hlist_empty(&inode->i_dentry))
  3196. goto out_unlock;
  3197. if (S_ISDIR(inode->i_mode)) {
  3198. alias = hlist_entry(inode->i_dentry.first, struct dentry, d_u.d_alias);
  3199. if (!IS_ROOT(alias))
  3200. dn = dget(alias);
  3201. goto out_unlock;
  3202. }
  3203. hlist_for_each_entry(alias, &inode->i_dentry, d_u.d_alias) {
  3204. spin_lock(&alias->d_lock);
  3205. if (!d_unhashed(alias) &&
  3206. (ceph_dentry(alias)->flags & CEPH_DENTRY_PRIMARY_LINK)) {
  3207. dn = dget_dlock(alias);
  3208. }
  3209. spin_unlock(&alias->d_lock);
  3210. if (dn)
  3211. break;
  3212. }
  3213. out_unlock:
  3214. spin_unlock(&inode->i_lock);
  3215. return dn;
  3216. }
  3217. /*
  3218. * Encode information about a cap for a reconnect with the MDS.
  3219. */
  3220. static int reconnect_caps_cb(struct inode *inode, struct ceph_cap *cap,
  3221. void *arg)
  3222. {
  3223. union {
  3224. struct ceph_mds_cap_reconnect v2;
  3225. struct ceph_mds_cap_reconnect_v1 v1;
  3226. } rec;
  3227. struct ceph_inode_info *ci = cap->ci;
  3228. struct ceph_reconnect_state *recon_state = arg;
  3229. struct ceph_pagelist *pagelist = recon_state->pagelist;
  3230. struct dentry *dentry;
  3231. char *path;
  3232. int pathlen = 0, err;
  3233. u64 pathbase;
  3234. u64 snap_follows;
  3235. dout(" adding %p ino %llx.%llx cap %p %lld %s\n",
  3236. inode, ceph_vinop(inode), cap, cap->cap_id,
  3237. ceph_cap_string(cap->issued));
  3238. dentry = d_find_primary(inode);
  3239. if (dentry) {
  3240. /* set pathbase to parent dir when msg_version >= 2 */
  3241. path = ceph_mdsc_build_path(dentry, &pathlen, &pathbase,
  3242. recon_state->msg_version >= 2);
  3243. dput(dentry);
  3244. if (IS_ERR(path)) {
  3245. err = PTR_ERR(path);
  3246. goto out_err;
  3247. }
  3248. } else {
  3249. path = NULL;
  3250. pathbase = 0;
  3251. }
  3252. spin_lock(&ci->i_ceph_lock);
  3253. cap->seq = 0; /* reset cap seq */
  3254. cap->issue_seq = 0; /* and issue_seq */
  3255. cap->mseq = 0; /* and migrate_seq */
  3256. cap->cap_gen = cap->session->s_cap_gen;
  3257. /* These are lost when the session goes away */
  3258. if (S_ISDIR(inode->i_mode)) {
  3259. if (cap->issued & CEPH_CAP_DIR_CREATE) {
  3260. ceph_put_string(rcu_dereference_raw(ci->i_cached_layout.pool_ns));
  3261. memset(&ci->i_cached_layout, 0, sizeof(ci->i_cached_layout));
  3262. }
  3263. cap->issued &= ~CEPH_CAP_ANY_DIR_OPS;
  3264. }
  3265. if (recon_state->msg_version >= 2) {
  3266. rec.v2.cap_id = cpu_to_le64(cap->cap_id);
  3267. rec.v2.wanted = cpu_to_le32(__ceph_caps_wanted(ci));
  3268. rec.v2.issued = cpu_to_le32(cap->issued);
  3269. rec.v2.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);
  3270. rec.v2.pathbase = cpu_to_le64(pathbase);
  3271. rec.v2.flock_len = (__force __le32)
  3272. ((ci->i_ceph_flags & CEPH_I_ERROR_FILELOCK) ? 0 : 1);
  3273. } else {
  3274. rec.v1.cap_id = cpu_to_le64(cap->cap_id);
  3275. rec.v1.wanted = cpu_to_le32(__ceph_caps_wanted(ci));
  3276. rec.v1.issued = cpu_to_le32(cap->issued);
  3277. rec.v1.size = cpu_to_le64(inode->i_size);
  3278. ceph_encode_timespec64(&rec.v1.mtime, &inode->i_mtime);
  3279. ceph_encode_timespec64(&rec.v1.atime, &inode->i_atime);
  3280. rec.v1.snaprealm = cpu_to_le64(ci->i_snap_realm->ino);
  3281. rec.v1.pathbase = cpu_to_le64(pathbase);
  3282. }
  3283. if (list_empty(&ci->i_cap_snaps)) {
  3284. snap_follows = ci->i_head_snapc ? ci->i_head_snapc->seq : 0;
  3285. } else {
  3286. struct ceph_cap_snap *capsnap =
  3287. list_first_entry(&ci->i_cap_snaps,
  3288. struct ceph_cap_snap, ci_item);
  3289. snap_follows = capsnap->follows;
  3290. }
  3291. spin_unlock(&ci->i_ceph_lock);
  3292. if (recon_state->msg_version >= 2) {
  3293. int num_fcntl_locks, num_flock_locks;
  3294. struct ceph_filelock *flocks = NULL;
  3295. size_t struct_len, total_len = sizeof(u64);
  3296. u8 struct_v = 0;
  3297. encode_again:
  3298. if (rec.v2.flock_len) {
  3299. ceph_count_locks(inode, &num_fcntl_locks, &num_flock_locks);
  3300. } else {
  3301. num_fcntl_locks = 0;
  3302. num_flock_locks = 0;
  3303. }
  3304. if (num_fcntl_locks + num_flock_locks > 0) {
  3305. flocks = kmalloc_array(num_fcntl_locks + num_flock_locks,
  3306. sizeof(struct ceph_filelock),
  3307. GFP_NOFS);
  3308. if (!flocks) {
  3309. err = -ENOMEM;
  3310. goto out_err;
  3311. }
  3312. err = ceph_encode_locks_to_buffer(inode, flocks,
  3313. num_fcntl_locks,
  3314. num_flock_locks);
  3315. if (err) {
  3316. kfree(flocks);
  3317. flocks = NULL;
  3318. if (err == -ENOSPC)
  3319. goto encode_again;
  3320. goto out_err;
  3321. }
  3322. } else {
  3323. kfree(flocks);
  3324. flocks = NULL;
  3325. }
  3326. if (recon_state->msg_version >= 3) {
  3327. /* version, compat_version and struct_len */
  3328. total_len += 2 * sizeof(u8) + sizeof(u32);
  3329. struct_v = 2;
  3330. }
  3331. /*
  3332. * number of encoded locks is stable, so copy to pagelist
  3333. */
  3334. struct_len = 2 * sizeof(u32) +
  3335. (num_fcntl_locks + num_flock_locks) *
  3336. sizeof(struct ceph_filelock);
  3337. rec.v2.flock_len = cpu_to_le32(struct_len);
  3338. struct_len += sizeof(u32) + pathlen + sizeof(rec.v2);
  3339. if (struct_v >= 2)
  3340. struct_len += sizeof(u64); /* snap_follows */
  3341. total_len += struct_len;
  3342. if (pagelist->length + total_len > RECONNECT_MAX_SIZE) {
  3343. err = send_reconnect_partial(recon_state);
  3344. if (err)
  3345. goto out_freeflocks;
  3346. pagelist = recon_state->pagelist;
  3347. }
  3348. err = ceph_pagelist_reserve(pagelist, total_len);
  3349. if (err)
  3350. goto out_freeflocks;
  3351. ceph_pagelist_encode_64(pagelist, ceph_ino(inode));
  3352. if (recon_state->msg_version >= 3) {
  3353. ceph_pagelist_encode_8(pagelist, struct_v);
  3354. ceph_pagelist_encode_8(pagelist, 1);
  3355. ceph_pagelist_encode_32(pagelist, struct_len);
  3356. }
  3357. ceph_pagelist_encode_string(pagelist, path, pathlen);
  3358. ceph_pagelist_append(pagelist, &rec, sizeof(rec.v2));
  3359. ceph_locks_to_pagelist(flocks, pagelist,
  3360. num_fcntl_locks, num_flock_locks);
  3361. if (struct_v >= 2)
  3362. ceph_pagelist_encode_64(pagelist, snap_follows);
  3363. out_freeflocks:
  3364. kfree(flocks);
  3365. } else {
  3366. err = ceph_pagelist_reserve(pagelist,
  3367. sizeof(u64) + sizeof(u32) +
  3368. pathlen + sizeof(rec.v1));
  3369. if (err)
  3370. goto out_err;
  3371. ceph_pagelist_encode_64(pagelist, ceph_ino(inode));
  3372. ceph_pagelist_encode_string(pagelist, path, pathlen);
  3373. ceph_pagelist_append(pagelist, &rec, sizeof(rec.v1));
  3374. }
  3375. out_err:
  3376. ceph_mdsc_free_path(path, pathlen);
  3377. if (!err)
  3378. recon_state->nr_caps++;
  3379. return err;
  3380. }
  3381. static int encode_snap_realms(struct ceph_mds_client *mdsc,
  3382. struct ceph_reconnect_state *recon_state)
  3383. {
  3384. struct rb_node *p;
  3385. struct ceph_pagelist *pagelist = recon_state->pagelist;
  3386. int err = 0;
  3387. if (recon_state->msg_version >= 4) {
  3388. err = ceph_pagelist_encode_32(pagelist, mdsc->num_snap_realms);
  3389. if (err < 0)
  3390. goto fail;
  3391. }
  3392. /*
  3393. * snaprealms. we provide mds with the ino, seq (version), and
  3394. * parent for all of our realms. If the mds has any newer info,
  3395. * it will tell us.
  3396. */
  3397. for (p = rb_first(&mdsc->snap_realms); p; p = rb_next(p)) {
  3398. struct ceph_snap_realm *realm =
  3399. rb_entry(p, struct ceph_snap_realm, node);
  3400. struct ceph_mds_snaprealm_reconnect sr_rec;
  3401. if (recon_state->msg_version >= 4) {
  3402. size_t need = sizeof(u8) * 2 + sizeof(u32) +
  3403. sizeof(sr_rec);
  3404. if (pagelist->length + need > RECONNECT_MAX_SIZE) {
  3405. err = send_reconnect_partial(recon_state);
  3406. if (err)
  3407. goto fail;
  3408. pagelist = recon_state->pagelist;
  3409. }
  3410. err = ceph_pagelist_reserve(pagelist, need);
  3411. if (err)
  3412. goto fail;
  3413. ceph_pagelist_encode_8(pagelist, 1);
  3414. ceph_pagelist_encode_8(pagelist, 1);
  3415. ceph_pagelist_encode_32(pagelist, sizeof(sr_rec));
  3416. }
  3417. dout(" adding snap realm %llx seq %lld parent %llx\n",
  3418. realm->ino, realm->seq, realm->parent_ino);
  3419. sr_rec.ino = cpu_to_le64(realm->ino);
  3420. sr_rec.seq = cpu_to_le64(realm->seq);
  3421. sr_rec.parent = cpu_to_le64(realm->parent_ino);
  3422. err = ceph_pagelist_append(pagelist, &sr_rec, sizeof(sr_rec));
  3423. if (err)
  3424. goto fail;
  3425. recon_state->nr_realms++;
  3426. }
  3427. fail:
  3428. return err;
  3429. }
  3430. /*
  3431. * If an MDS fails and recovers, clients need to reconnect in order to
  3432. * reestablish shared state. This includes all caps issued through
  3433. * this session _and_ the snap_realm hierarchy. Because it's not
  3434. * clear which snap realms the mds cares about, we send everything we
  3435. * know about.. that ensures we'll then get any new info the
  3436. * recovering MDS might have.
  3437. *
  3438. * This is a relatively heavyweight operation, but it's rare.
  3439. */
  3440. static void send_mds_reconnect(struct ceph_mds_client *mdsc,
  3441. struct ceph_mds_session *session)
  3442. {
  3443. struct ceph_msg *reply;
  3444. int mds = session->s_mds;
  3445. int err = -ENOMEM;
  3446. struct ceph_reconnect_state recon_state = {
  3447. .session = session,
  3448. };
  3449. LIST_HEAD(dispose);
  3450. pr_info("mds%d reconnect start\n", mds);
  3451. recon_state.pagelist = ceph_pagelist_alloc(GFP_NOFS);
  3452. if (!recon_state.pagelist)
  3453. goto fail_nopagelist;
  3454. reply = ceph_msg_new2(CEPH_MSG_CLIENT_RECONNECT, 0, 1, GFP_NOFS, false);
  3455. if (!reply)
  3456. goto fail_nomsg;
  3457. xa_destroy(&session->s_delegated_inos);
  3458. mutex_lock(&session->s_mutex);
  3459. session->s_state = CEPH_MDS_SESSION_RECONNECTING;
  3460. session->s_seq = 0;
  3461. dout("session %p state %s\n", session,
  3462. ceph_session_state_name(session->s_state));
  3463. spin_lock(&session->s_gen_ttl_lock);
  3464. session->s_cap_gen++;
  3465. spin_unlock(&session->s_gen_ttl_lock);
  3466. spin_lock(&session->s_cap_lock);
  3467. /* don't know if session is readonly */
  3468. session->s_readonly = 0;
  3469. /*
  3470. * notify __ceph_remove_cap() that we are composing cap reconnect.
  3471. * If a cap get released before being added to the cap reconnect,
  3472. * __ceph_remove_cap() should skip queuing cap release.
  3473. */
  3474. session->s_cap_reconnect = 1;
  3475. /* drop old cap expires; we're about to reestablish that state */
  3476. detach_cap_releases(session, &dispose);
  3477. spin_unlock(&session->s_cap_lock);
  3478. dispose_cap_releases(mdsc, &dispose);
  3479. /* trim unused caps to reduce MDS's cache rejoin time */
  3480. if (mdsc->fsc->sb->s_root)
  3481. shrink_dcache_parent(mdsc->fsc->sb->s_root);
  3482. ceph_con_close(&session->s_con);
  3483. ceph_con_open(&session->s_con,
  3484. CEPH_ENTITY_TYPE_MDS, mds,
  3485. ceph_mdsmap_get_addr(mdsc->mdsmap, mds));
  3486. /* replay unsafe requests */
  3487. replay_unsafe_requests(mdsc, session);
  3488. ceph_early_kick_flushing_caps(mdsc, session);
  3489. down_read(&mdsc->snap_rwsem);
  3490. /* placeholder for nr_caps */
  3491. err = ceph_pagelist_encode_32(recon_state.pagelist, 0);
  3492. if (err)
  3493. goto fail;
  3494. if (test_bit(CEPHFS_FEATURE_MULTI_RECONNECT, &session->s_features)) {
  3495. recon_state.msg_version = 3;
  3496. recon_state.allow_multi = true;
  3497. } else if (session->s_con.peer_features & CEPH_FEATURE_MDSENC) {
  3498. recon_state.msg_version = 3;
  3499. } else {
  3500. recon_state.msg_version = 2;
  3501. }
  3502. /* trsaverse this session's caps */
  3503. err = ceph_iterate_session_caps(session, reconnect_caps_cb, &recon_state);
  3504. spin_lock(&session->s_cap_lock);
  3505. session->s_cap_reconnect = 0;
  3506. spin_unlock(&session->s_cap_lock);
  3507. if (err < 0)
  3508. goto fail;
  3509. /* check if all realms can be encoded into current message */
  3510. if (mdsc->num_snap_realms) {
  3511. size_t total_len =
  3512. recon_state.pagelist->length +
  3513. mdsc->num_snap_realms *
  3514. sizeof(struct ceph_mds_snaprealm_reconnect);
  3515. if (recon_state.msg_version >= 4) {
  3516. /* number of realms */
  3517. total_len += sizeof(u32);
  3518. /* version, compat_version and struct_len */
  3519. total_len += mdsc->num_snap_realms *
  3520. (2 * sizeof(u8) + sizeof(u32));
  3521. }
  3522. if (total_len > RECONNECT_MAX_SIZE) {
  3523. if (!recon_state.allow_multi) {
  3524. err = -ENOSPC;
  3525. goto fail;
  3526. }
  3527. if (recon_state.nr_caps) {
  3528. err = send_reconnect_partial(&recon_state);
  3529. if (err)
  3530. goto fail;
  3531. }
  3532. recon_state.msg_version = 5;
  3533. }
  3534. }
  3535. err = encode_snap_realms(mdsc, &recon_state);
  3536. if (err < 0)
  3537. goto fail;
  3538. if (recon_state.msg_version >= 5) {
  3539. err = ceph_pagelist_encode_8(recon_state.pagelist, 0);
  3540. if (err < 0)
  3541. goto fail;
  3542. }
  3543. if (recon_state.nr_caps || recon_state.nr_realms) {
  3544. struct page *page =
  3545. list_first_entry(&recon_state.pagelist->head,
  3546. struct page, lru);
  3547. __le32 *addr = kmap_atomic(page);
  3548. if (recon_state.nr_caps) {
  3549. WARN_ON(recon_state.nr_realms != mdsc->num_snap_realms);
  3550. *addr = cpu_to_le32(recon_state.nr_caps);
  3551. } else if (recon_state.msg_version >= 4) {
  3552. *(addr + 1) = cpu_to_le32(recon_state.nr_realms);
  3553. }
  3554. kunmap_atomic(addr);
  3555. }
  3556. reply->hdr.version = cpu_to_le16(recon_state.msg_version);
  3557. if (recon_state.msg_version >= 4)
  3558. reply->hdr.compat_version = cpu_to_le16(4);
  3559. reply->hdr.data_len = cpu_to_le32(recon_state.pagelist->length);
  3560. ceph_msg_data_add_pagelist(reply, recon_state.pagelist);
  3561. ceph_con_send(&session->s_con, reply);
  3562. mutex_unlock(&session->s_mutex);
  3563. mutex_lock(&mdsc->mutex);
  3564. __wake_requests(mdsc, &session->s_waiting);
  3565. mutex_unlock(&mdsc->mutex);
  3566. up_read(&mdsc->snap_rwsem);
  3567. ceph_pagelist_release(recon_state.pagelist);
  3568. return;
  3569. fail:
  3570. ceph_msg_put(reply);
  3571. up_read(&mdsc->snap_rwsem);
  3572. mutex_unlock(&session->s_mutex);
  3573. fail_nomsg:
  3574. ceph_pagelist_release(recon_state.pagelist);
  3575. fail_nopagelist:
  3576. pr_err("error %d preparing reconnect for mds%d\n", err, mds);
  3577. return;
  3578. }
  3579. /*
  3580. * compare old and new mdsmaps, kicking requests
  3581. * and closing out old connections as necessary
  3582. *
  3583. * called under mdsc->mutex.
  3584. */
  3585. static void check_new_map(struct ceph_mds_client *mdsc,
  3586. struct ceph_mdsmap *newmap,
  3587. struct ceph_mdsmap *oldmap)
  3588. {
  3589. int i;
  3590. int oldstate, newstate;
  3591. struct ceph_mds_session *s;
  3592. dout("check_new_map new %u old %u\n",
  3593. newmap->m_epoch, oldmap->m_epoch);
  3594. for (i = 0; i < oldmap->possible_max_rank && i < mdsc->max_sessions; i++) {
  3595. if (!mdsc->sessions[i])
  3596. continue;
  3597. s = mdsc->sessions[i];
  3598. oldstate = ceph_mdsmap_get_state(oldmap, i);
  3599. newstate = ceph_mdsmap_get_state(newmap, i);
  3600. dout("check_new_map mds%d state %s%s -> %s%s (session %s)\n",
  3601. i, ceph_mds_state_name(oldstate),
  3602. ceph_mdsmap_is_laggy(oldmap, i) ? " (laggy)" : "",
  3603. ceph_mds_state_name(newstate),
  3604. ceph_mdsmap_is_laggy(newmap, i) ? " (laggy)" : "",
  3605. ceph_session_state_name(s->s_state));
  3606. if (i >= newmap->possible_max_rank) {
  3607. /* force close session for stopped mds */
  3608. ceph_get_mds_session(s);
  3609. __unregister_session(mdsc, s);
  3610. __wake_requests(mdsc, &s->s_waiting);
  3611. mutex_unlock(&mdsc->mutex);
  3612. mutex_lock(&s->s_mutex);
  3613. cleanup_session_requests(mdsc, s);
  3614. remove_session_caps(s);
  3615. mutex_unlock(&s->s_mutex);
  3616. ceph_put_mds_session(s);
  3617. mutex_lock(&mdsc->mutex);
  3618. kick_requests(mdsc, i);
  3619. continue;
  3620. }
  3621. if (memcmp(ceph_mdsmap_get_addr(oldmap, i),
  3622. ceph_mdsmap_get_addr(newmap, i),
  3623. sizeof(struct ceph_entity_addr))) {
  3624. /* just close it */
  3625. mutex_unlock(&mdsc->mutex);
  3626. mutex_lock(&s->s_mutex);
  3627. mutex_lock(&mdsc->mutex);
  3628. ceph_con_close(&s->s_con);
  3629. mutex_unlock(&s->s_mutex);
  3630. s->s_state = CEPH_MDS_SESSION_RESTARTING;
  3631. } else if (oldstate == newstate) {
  3632. continue; /* nothing new with this mds */
  3633. }
  3634. /*
  3635. * send reconnect?
  3636. */
  3637. if (s->s_state == CEPH_MDS_SESSION_RESTARTING &&
  3638. newstate >= CEPH_MDS_STATE_RECONNECT) {
  3639. mutex_unlock(&mdsc->mutex);
  3640. send_mds_reconnect(mdsc, s);
  3641. mutex_lock(&mdsc->mutex);
  3642. }
  3643. /*
  3644. * kick request on any mds that has gone active.
  3645. */
  3646. if (oldstate < CEPH_MDS_STATE_ACTIVE &&
  3647. newstate >= CEPH_MDS_STATE_ACTIVE) {
  3648. if (oldstate != CEPH_MDS_STATE_CREATING &&
  3649. oldstate != CEPH_MDS_STATE_STARTING)
  3650. pr_info("mds%d recovery completed\n", s->s_mds);
  3651. kick_requests(mdsc, i);
  3652. mutex_unlock(&mdsc->mutex);
  3653. mutex_lock(&s->s_mutex);
  3654. mutex_lock(&mdsc->mutex);
  3655. ceph_kick_flushing_caps(mdsc, s);
  3656. mutex_unlock(&s->s_mutex);
  3657. wake_up_session_caps(s, RECONNECT);
  3658. }
  3659. }
  3660. for (i = 0; i < newmap->possible_max_rank && i < mdsc->max_sessions; i++) {
  3661. s = mdsc->sessions[i];
  3662. if (!s)
  3663. continue;
  3664. if (!ceph_mdsmap_is_laggy(newmap, i))
  3665. continue;
  3666. if (s->s_state == CEPH_MDS_SESSION_OPEN ||
  3667. s->s_state == CEPH_MDS_SESSION_HUNG ||
  3668. s->s_state == CEPH_MDS_SESSION_CLOSING) {
  3669. dout(" connecting to export targets of laggy mds%d\n",
  3670. i);
  3671. __open_export_target_sessions(mdsc, s);
  3672. }
  3673. }
  3674. }
  3675. /*
  3676. * leases
  3677. */
  3678. /*
  3679. * caller must hold session s_mutex, dentry->d_lock
  3680. */
  3681. void __ceph_mdsc_drop_dentry_lease(struct dentry *dentry)
  3682. {
  3683. struct ceph_dentry_info *di = ceph_dentry(dentry);
  3684. ceph_put_mds_session(di->lease_session);
  3685. di->lease_session = NULL;
  3686. }
  3687. static void handle_lease(struct ceph_mds_client *mdsc,
  3688. struct ceph_mds_session *session,
  3689. struct ceph_msg *msg)
  3690. {
  3691. struct super_block *sb = mdsc->fsc->sb;
  3692. struct inode *inode;
  3693. struct dentry *parent, *dentry;
  3694. struct ceph_dentry_info *di;
  3695. int mds = session->s_mds;
  3696. struct ceph_mds_lease *h = msg->front.iov_base;
  3697. u32 seq;
  3698. struct ceph_vino vino;
  3699. struct qstr dname;
  3700. int release = 0;
  3701. dout("handle_lease from mds%d\n", mds);
  3702. /* decode */
  3703. if (msg->front.iov_len < sizeof(*h) + sizeof(u32))
  3704. goto bad;
  3705. vino.ino = le64_to_cpu(h->ino);
  3706. vino.snap = CEPH_NOSNAP;
  3707. seq = le32_to_cpu(h->seq);
  3708. dname.len = get_unaligned_le32(h + 1);
  3709. if (msg->front.iov_len < sizeof(*h) + sizeof(u32) + dname.len)
  3710. goto bad;
  3711. dname.name = (void *)(h + 1) + sizeof(u32);
  3712. /* lookup inode */
  3713. inode = ceph_find_inode(sb, vino);
  3714. dout("handle_lease %s, ino %llx %p %.*s\n",
  3715. ceph_lease_op_name(h->action), vino.ino, inode,
  3716. dname.len, dname.name);
  3717. mutex_lock(&session->s_mutex);
  3718. inc_session_sequence(session);
  3719. if (!inode) {
  3720. dout("handle_lease no inode %llx\n", vino.ino);
  3721. goto release;
  3722. }
  3723. /* dentry */
  3724. parent = d_find_alias(inode);
  3725. if (!parent) {
  3726. dout("no parent dentry on inode %p\n", inode);
  3727. WARN_ON(1);
  3728. goto release; /* hrm... */
  3729. }
  3730. dname.hash = full_name_hash(parent, dname.name, dname.len);
  3731. dentry = d_lookup(parent, &dname);
  3732. dput(parent);
  3733. if (!dentry)
  3734. goto release;
  3735. spin_lock(&dentry->d_lock);
  3736. di = ceph_dentry(dentry);
  3737. switch (h->action) {
  3738. case CEPH_MDS_LEASE_REVOKE:
  3739. if (di->lease_session == session) {
  3740. if (ceph_seq_cmp(di->lease_seq, seq) > 0)
  3741. h->seq = cpu_to_le32(di->lease_seq);
  3742. __ceph_mdsc_drop_dentry_lease(dentry);
  3743. }
  3744. release = 1;
  3745. break;
  3746. case CEPH_MDS_LEASE_RENEW:
  3747. if (di->lease_session == session &&
  3748. di->lease_gen == session->s_cap_gen &&
  3749. di->lease_renew_from &&
  3750. di->lease_renew_after == 0) {
  3751. unsigned long duration =
  3752. msecs_to_jiffies(le32_to_cpu(h->duration_ms));
  3753. di->lease_seq = seq;
  3754. di->time = di->lease_renew_from + duration;
  3755. di->lease_renew_after = di->lease_renew_from +
  3756. (duration >> 1);
  3757. di->lease_renew_from = 0;
  3758. }
  3759. break;
  3760. }
  3761. spin_unlock(&dentry->d_lock);
  3762. dput(dentry);
  3763. if (!release)
  3764. goto out;
  3765. release:
  3766. /* let's just reuse the same message */
  3767. h->action = CEPH_MDS_LEASE_REVOKE_ACK;
  3768. ceph_msg_get(msg);
  3769. ceph_con_send(&session->s_con, msg);
  3770. out:
  3771. mutex_unlock(&session->s_mutex);
  3772. /* avoid calling iput_final() in mds dispatch threads */
  3773. ceph_async_iput(inode);
  3774. return;
  3775. bad:
  3776. pr_err("corrupt lease message\n");
  3777. ceph_msg_dump(msg);
  3778. }
  3779. void ceph_mdsc_lease_send_msg(struct ceph_mds_session *session,
  3780. struct dentry *dentry, char action,
  3781. u32 seq)
  3782. {
  3783. struct ceph_msg *msg;
  3784. struct ceph_mds_lease *lease;
  3785. struct inode *dir;
  3786. int len = sizeof(*lease) + sizeof(u32) + NAME_MAX;
  3787. dout("lease_send_msg identry %p %s to mds%d\n",
  3788. dentry, ceph_lease_op_name(action), session->s_mds);
  3789. msg = ceph_msg_new(CEPH_MSG_CLIENT_LEASE, len, GFP_NOFS, false);
  3790. if (!msg)
  3791. return;
  3792. lease = msg->front.iov_base;
  3793. lease->action = action;
  3794. lease->seq = cpu_to_le32(seq);
  3795. spin_lock(&dentry->d_lock);
  3796. dir = d_inode(dentry->d_parent);
  3797. lease->ino = cpu_to_le64(ceph_ino(dir));
  3798. lease->first = lease->last = cpu_to_le64(ceph_snap(dir));
  3799. put_unaligned_le32(dentry->d_name.len, lease + 1);
  3800. memcpy((void *)(lease + 1) + 4,
  3801. dentry->d_name.name, dentry->d_name.len);
  3802. spin_unlock(&dentry->d_lock);
  3803. /*
  3804. * if this is a preemptive lease RELEASE, no need to
  3805. * flush request stream, since the actual request will
  3806. * soon follow.
  3807. */
  3808. msg->more_to_follow = (action == CEPH_MDS_LEASE_RELEASE);
  3809. ceph_con_send(&session->s_con, msg);
  3810. }
  3811. /*
  3812. * lock unlock sessions, to wait ongoing session activities
  3813. */
  3814. static void lock_unlock_sessions(struct ceph_mds_client *mdsc)
  3815. {
  3816. int i;
  3817. mutex_lock(&mdsc->mutex);
  3818. for (i = 0; i < mdsc->max_sessions; i++) {
  3819. struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);
  3820. if (!s)
  3821. continue;
  3822. mutex_unlock(&mdsc->mutex);
  3823. mutex_lock(&s->s_mutex);
  3824. mutex_unlock(&s->s_mutex);
  3825. ceph_put_mds_session(s);
  3826. mutex_lock(&mdsc->mutex);
  3827. }
  3828. mutex_unlock(&mdsc->mutex);
  3829. }
  3830. static void maybe_recover_session(struct ceph_mds_client *mdsc)
  3831. {
  3832. struct ceph_fs_client *fsc = mdsc->fsc;
  3833. if (!ceph_test_mount_opt(fsc, CLEANRECOVER))
  3834. return;
  3835. if (READ_ONCE(fsc->mount_state) != CEPH_MOUNT_MOUNTED)
  3836. return;
  3837. if (!READ_ONCE(fsc->blocklisted))
  3838. return;
  3839. if (fsc->last_auto_reconnect &&
  3840. time_before(jiffies, fsc->last_auto_reconnect + HZ * 60 * 30))
  3841. return;
  3842. pr_info("auto reconnect after blocklisted\n");
  3843. fsc->last_auto_reconnect = jiffies;
  3844. ceph_force_reconnect(fsc->sb);
  3845. }
  3846. bool check_session_state(struct ceph_mds_session *s)
  3847. {
  3848. switch (s->s_state) {
  3849. case CEPH_MDS_SESSION_OPEN:
  3850. if (s->s_ttl && time_after(jiffies, s->s_ttl)) {
  3851. s->s_state = CEPH_MDS_SESSION_HUNG;
  3852. pr_info("mds%d hung\n", s->s_mds);
  3853. }
  3854. break;
  3855. case CEPH_MDS_SESSION_CLOSING:
  3856. /* Should never reach this when we're unmounting */
  3857. WARN_ON_ONCE(s->s_ttl);
  3858. fallthrough;
  3859. case CEPH_MDS_SESSION_NEW:
  3860. case CEPH_MDS_SESSION_RESTARTING:
  3861. case CEPH_MDS_SESSION_CLOSED:
  3862. case CEPH_MDS_SESSION_REJECTED:
  3863. return false;
  3864. }
  3865. return true;
  3866. }
  3867. /*
  3868. * If the sequence is incremented while we're waiting on a REQUEST_CLOSE reply,
  3869. * then we need to retransmit that request.
  3870. */
  3871. void inc_session_sequence(struct ceph_mds_session *s)
  3872. {
  3873. lockdep_assert_held(&s->s_mutex);
  3874. s->s_seq++;
  3875. if (s->s_state == CEPH_MDS_SESSION_CLOSING) {
  3876. int ret;
  3877. dout("resending session close request for mds%d\n", s->s_mds);
  3878. ret = request_close_session(s);
  3879. if (ret < 0)
  3880. pr_err("unable to close session to mds%d: %d\n",
  3881. s->s_mds, ret);
  3882. }
  3883. }
  3884. /*
  3885. * delayed work -- periodically trim expired leases, renew caps with mds. If
  3886. * the @delay parameter is set to 0 or if it's more than 5 secs, the default
  3887. * workqueue delay value of 5 secs will be used.
  3888. */
  3889. static void schedule_delayed(struct ceph_mds_client *mdsc, unsigned long delay)
  3890. {
  3891. unsigned long max_delay = HZ * 5;
  3892. /* 5 secs default delay */
  3893. if (!delay || (delay > max_delay))
  3894. delay = max_delay;
  3895. schedule_delayed_work(&mdsc->delayed_work,
  3896. round_jiffies_relative(delay));
  3897. }
  3898. static void delayed_work(struct work_struct *work)
  3899. {
  3900. struct ceph_mds_client *mdsc =
  3901. container_of(work, struct ceph_mds_client, delayed_work.work);
  3902. unsigned long delay;
  3903. int renew_interval;
  3904. int renew_caps;
  3905. int i;
  3906. dout("mdsc delayed_work\n");
  3907. if (mdsc->stopping)
  3908. return;
  3909. mutex_lock(&mdsc->mutex);
  3910. renew_interval = mdsc->mdsmap->m_session_timeout >> 2;
  3911. renew_caps = time_after_eq(jiffies, HZ*renew_interval +
  3912. mdsc->last_renew_caps);
  3913. if (renew_caps)
  3914. mdsc->last_renew_caps = jiffies;
  3915. for (i = 0; i < mdsc->max_sessions; i++) {
  3916. struct ceph_mds_session *s = __ceph_lookup_mds_session(mdsc, i);
  3917. if (!s)
  3918. continue;
  3919. if (!check_session_state(s)) {
  3920. ceph_put_mds_session(s);
  3921. continue;
  3922. }
  3923. mutex_unlock(&mdsc->mutex);
  3924. mutex_lock(&s->s_mutex);
  3925. if (renew_caps)
  3926. send_renew_caps(mdsc, s);
  3927. else
  3928. ceph_con_keepalive(&s->s_con);
  3929. if (s->s_state == CEPH_MDS_SESSION_OPEN ||
  3930. s->s_state == CEPH_MDS_SESSION_HUNG)
  3931. ceph_send_cap_releases(mdsc, s);
  3932. mutex_unlock(&s->s_mutex);
  3933. ceph_put_mds_session(s);
  3934. mutex_lock(&mdsc->mutex);
  3935. }
  3936. mutex_unlock(&mdsc->mutex);
  3937. delay = ceph_check_delayed_caps(mdsc);
  3938. ceph_queue_cap_reclaim_work(mdsc);
  3939. ceph_trim_snapid_map(mdsc);
  3940. maybe_recover_session(mdsc);
  3941. schedule_delayed(mdsc, delay);
  3942. }
  3943. int ceph_mdsc_init(struct ceph_fs_client *fsc)
  3944. {
  3945. struct ceph_mds_client *mdsc;
  3946. int err;
  3947. mdsc = kzalloc(sizeof(struct ceph_mds_client), GFP_NOFS);
  3948. if (!mdsc)
  3949. return -ENOMEM;
  3950. mdsc->fsc = fsc;
  3951. mutex_init(&mdsc->mutex);
  3952. mdsc->mdsmap = kzalloc(sizeof(*mdsc->mdsmap), GFP_NOFS);
  3953. if (!mdsc->mdsmap) {
  3954. err = -ENOMEM;
  3955. goto err_mdsc;
  3956. }
  3957. init_completion(&mdsc->safe_umount_waiters);
  3958. init_waitqueue_head(&mdsc->session_close_wq);
  3959. INIT_LIST_HEAD(&mdsc->waiting_for_map);
  3960. mdsc->sessions = NULL;
  3961. atomic_set(&mdsc->num_sessions, 0);
  3962. mdsc->max_sessions = 0;
  3963. mdsc->stopping = 0;
  3964. atomic64_set(&mdsc->quotarealms_count, 0);
  3965. mdsc->quotarealms_inodes = RB_ROOT;
  3966. mutex_init(&mdsc->quotarealms_inodes_mutex);
  3967. mdsc->last_snap_seq = 0;
  3968. init_rwsem(&mdsc->snap_rwsem);
  3969. mdsc->snap_realms = RB_ROOT;
  3970. INIT_LIST_HEAD(&mdsc->snap_empty);
  3971. mdsc->num_snap_realms = 0;
  3972. spin_lock_init(&mdsc->snap_empty_lock);
  3973. mdsc->last_tid = 0;
  3974. mdsc->oldest_tid = 0;
  3975. mdsc->request_tree = RB_ROOT;
  3976. INIT_DELAYED_WORK(&mdsc->delayed_work, delayed_work);
  3977. mdsc->last_renew_caps = jiffies;
  3978. INIT_LIST_HEAD(&mdsc->cap_delay_list);
  3979. INIT_LIST_HEAD(&mdsc->cap_wait_list);
  3980. spin_lock_init(&mdsc->cap_delay_lock);
  3981. INIT_LIST_HEAD(&mdsc->snap_flush_list);
  3982. spin_lock_init(&mdsc->snap_flush_lock);
  3983. mdsc->last_cap_flush_tid = 1;
  3984. INIT_LIST_HEAD(&mdsc->cap_flush_list);
  3985. INIT_LIST_HEAD(&mdsc->cap_dirty_migrating);
  3986. mdsc->num_cap_flushing = 0;
  3987. spin_lock_init(&mdsc->cap_dirty_lock);
  3988. init_waitqueue_head(&mdsc->cap_flushing_wq);
  3989. INIT_WORK(&mdsc->cap_reclaim_work, ceph_cap_reclaim_work);
  3990. atomic_set(&mdsc->cap_reclaim_pending, 0);
  3991. err = ceph_metric_init(&mdsc->metric);
  3992. if (err)
  3993. goto err_mdsmap;
  3994. spin_lock_init(&mdsc->dentry_list_lock);
  3995. INIT_LIST_HEAD(&mdsc->dentry_leases);
  3996. INIT_LIST_HEAD(&mdsc->dentry_dir_leases);
  3997. ceph_caps_init(mdsc);
  3998. ceph_adjust_caps_max_min(mdsc, fsc->mount_options);
  3999. spin_lock_init(&mdsc->snapid_map_lock);
  4000. mdsc->snapid_map_tree = RB_ROOT;
  4001. INIT_LIST_HEAD(&mdsc->snapid_map_lru);
  4002. init_rwsem(&mdsc->pool_perm_rwsem);
  4003. mdsc->pool_perm_tree = RB_ROOT;
  4004. strscpy(mdsc->nodename, utsname()->nodename,
  4005. sizeof(mdsc->nodename));
  4006. fsc->mdsc = mdsc;
  4007. return 0;
  4008. err_mdsmap:
  4009. kfree(mdsc->mdsmap);
  4010. err_mdsc:
  4011. kfree(mdsc);
  4012. return err;
  4013. }
  4014. /*
  4015. * Wait for safe replies on open mds requests. If we time out, drop
  4016. * all requests from the tree to avoid dangling dentry refs.
  4017. */
  4018. static void wait_requests(struct ceph_mds_client *mdsc)
  4019. {
  4020. struct ceph_options *opts = mdsc->fsc->client->options;
  4021. struct ceph_mds_request *req;
  4022. mutex_lock(&mdsc->mutex);
  4023. if (__get_oldest_req(mdsc)) {
  4024. mutex_unlock(&mdsc->mutex);
  4025. dout("wait_requests waiting for requests\n");
  4026. wait_for_completion_timeout(&mdsc->safe_umount_waiters,
  4027. ceph_timeout_jiffies(opts->mount_timeout));
  4028. /* tear down remaining requests */
  4029. mutex_lock(&mdsc->mutex);
  4030. while ((req = __get_oldest_req(mdsc))) {
  4031. dout("wait_requests timed out on tid %llu\n",
  4032. req->r_tid);
  4033. list_del_init(&req->r_wait);
  4034. __unregister_request(mdsc, req);
  4035. }
  4036. }
  4037. mutex_unlock(&mdsc->mutex);
  4038. dout("wait_requests done\n");
  4039. }
  4040. /*
  4041. * called before mount is ro, and before dentries are torn down.
  4042. * (hmm, does this still race with new lookups?)
  4043. */
  4044. void ceph_mdsc_pre_umount(struct ceph_mds_client *mdsc)
  4045. {
  4046. dout("pre_umount\n");
  4047. mdsc->stopping = 1;
  4048. lock_unlock_sessions(mdsc);
  4049. ceph_flush_dirty_caps(mdsc);
  4050. wait_requests(mdsc);
  4051. /*
  4052. * wait for reply handlers to drop their request refs and
  4053. * their inode/dcache refs
  4054. */
  4055. ceph_msgr_flush();
  4056. ceph_cleanup_quotarealms_inodes(mdsc);
  4057. }
  4058. /*
  4059. * wait for all write mds requests to flush.
  4060. */
  4061. static void wait_unsafe_requests(struct ceph_mds_client *mdsc, u64 want_tid)
  4062. {
  4063. struct ceph_mds_request *req = NULL, *nextreq;
  4064. struct rb_node *n;
  4065. mutex_lock(&mdsc->mutex);
  4066. dout("wait_unsafe_requests want %lld\n", want_tid);
  4067. restart:
  4068. req = __get_oldest_req(mdsc);
  4069. while (req && req->r_tid <= want_tid) {
  4070. /* find next request */
  4071. n = rb_next(&req->r_node);
  4072. if (n)
  4073. nextreq = rb_entry(n, struct ceph_mds_request, r_node);
  4074. else
  4075. nextreq = NULL;
  4076. if (req->r_op != CEPH_MDS_OP_SETFILELOCK &&
  4077. (req->r_op & CEPH_MDS_OP_WRITE)) {
  4078. /* write op */
  4079. ceph_mdsc_get_request(req);
  4080. if (nextreq)
  4081. ceph_mdsc_get_request(nextreq);
  4082. mutex_unlock(&mdsc->mutex);
  4083. dout("wait_unsafe_requests wait on %llu (want %llu)\n",
  4084. req->r_tid, want_tid);
  4085. wait_for_completion(&req->r_safe_completion);
  4086. mutex_lock(&mdsc->mutex);
  4087. ceph_mdsc_put_request(req);
  4088. if (!nextreq)
  4089. break; /* next dne before, so we're done! */
  4090. if (RB_EMPTY_NODE(&nextreq->r_node)) {
  4091. /* next request was removed from tree */
  4092. ceph_mdsc_put_request(nextreq);
  4093. goto restart;
  4094. }
  4095. ceph_mdsc_put_request(nextreq); /* won't go away */
  4096. }
  4097. req = nextreq;
  4098. }
  4099. mutex_unlock(&mdsc->mutex);
  4100. dout("wait_unsafe_requests done\n");
  4101. }
  4102. void ceph_mdsc_sync(struct ceph_mds_client *mdsc)
  4103. {
  4104. u64 want_tid, want_flush;
  4105. if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN)
  4106. return;
  4107. dout("sync\n");
  4108. mutex_lock(&mdsc->mutex);
  4109. want_tid = mdsc->last_tid;
  4110. mutex_unlock(&mdsc->mutex);
  4111. ceph_flush_dirty_caps(mdsc);
  4112. spin_lock(&mdsc->cap_dirty_lock);
  4113. want_flush = mdsc->last_cap_flush_tid;
  4114. if (!list_empty(&mdsc->cap_flush_list)) {
  4115. struct ceph_cap_flush *cf =
  4116. list_last_entry(&mdsc->cap_flush_list,
  4117. struct ceph_cap_flush, g_list);
  4118. cf->wake = true;
  4119. }
  4120. spin_unlock(&mdsc->cap_dirty_lock);
  4121. dout("sync want tid %lld flush_seq %lld\n",
  4122. want_tid, want_flush);
  4123. wait_unsafe_requests(mdsc, want_tid);
  4124. wait_caps_flush(mdsc, want_flush);
  4125. }
  4126. /*
  4127. * true if all sessions are closed, or we force unmount
  4128. */
  4129. static bool done_closing_sessions(struct ceph_mds_client *mdsc, int skipped)
  4130. {
  4131. if (READ_ONCE(mdsc->fsc->mount_state) == CEPH_MOUNT_SHUTDOWN)
  4132. return true;
  4133. return atomic_read(&mdsc->num_sessions) <= skipped;
  4134. }
  4135. /*
  4136. * called after sb is ro.
  4137. */
  4138. void ceph_mdsc_close_sessions(struct ceph_mds_client *mdsc)
  4139. {
  4140. struct ceph_options *opts = mdsc->fsc->client->options;
  4141. struct ceph_mds_session *session;
  4142. int i;
  4143. int skipped = 0;
  4144. dout("close_sessions\n");
  4145. /* close sessions */
  4146. mutex_lock(&mdsc->mutex);
  4147. for (i = 0; i < mdsc->max_sessions; i++) {
  4148. session = __ceph_lookup_mds_session(mdsc, i);
  4149. if (!session)
  4150. continue;
  4151. mutex_unlock(&mdsc->mutex);
  4152. mutex_lock(&session->s_mutex);
  4153. if (__close_session(mdsc, session) <= 0)
  4154. skipped++;
  4155. mutex_unlock(&session->s_mutex);
  4156. ceph_put_mds_session(session);
  4157. mutex_lock(&mdsc->mutex);
  4158. }
  4159. mutex_unlock(&mdsc->mutex);
  4160. dout("waiting for sessions to close\n");
  4161. wait_event_timeout(mdsc->session_close_wq,
  4162. done_closing_sessions(mdsc, skipped),
  4163. ceph_timeout_jiffies(opts->mount_timeout));
  4164. /* tear down remaining sessions */
  4165. mutex_lock(&mdsc->mutex);
  4166. for (i = 0; i < mdsc->max_sessions; i++) {
  4167. if (mdsc->sessions[i]) {
  4168. session = ceph_get_mds_session(mdsc->sessions[i]);
  4169. __unregister_session(mdsc, session);
  4170. mutex_unlock(&mdsc->mutex);
  4171. mutex_lock(&session->s_mutex);
  4172. remove_session_caps(session);
  4173. mutex_unlock(&session->s_mutex);
  4174. ceph_put_mds_session(session);
  4175. mutex_lock(&mdsc->mutex);
  4176. }
  4177. }
  4178. WARN_ON(!list_empty(&mdsc->cap_delay_list));
  4179. mutex_unlock(&mdsc->mutex);
  4180. ceph_cleanup_snapid_map(mdsc);
  4181. ceph_cleanup_empty_realms(mdsc);
  4182. cancel_work_sync(&mdsc->cap_reclaim_work);
  4183. cancel_delayed_work_sync(&mdsc->delayed_work); /* cancel timer */
  4184. dout("stopped\n");
  4185. }
  4186. void ceph_mdsc_force_umount(struct ceph_mds_client *mdsc)
  4187. {
  4188. struct ceph_mds_session *session;
  4189. int mds;
  4190. dout("force umount\n");
  4191. mutex_lock(&mdsc->mutex);
  4192. for (mds = 0; mds < mdsc->max_sessions; mds++) {
  4193. session = __ceph_lookup_mds_session(mdsc, mds);
  4194. if (!session)
  4195. continue;
  4196. if (session->s_state == CEPH_MDS_SESSION_REJECTED)
  4197. __unregister_session(mdsc, session);
  4198. __wake_requests(mdsc, &session->s_waiting);
  4199. mutex_unlock(&mdsc->mutex);
  4200. mutex_lock(&session->s_mutex);
  4201. __close_session(mdsc, session);
  4202. if (session->s_state == CEPH_MDS_SESSION_CLOSING) {
  4203. cleanup_session_requests(mdsc, session);
  4204. remove_session_caps(session);
  4205. }
  4206. mutex_unlock(&session->s_mutex);
  4207. ceph_put_mds_session(session);
  4208. mutex_lock(&mdsc->mutex);
  4209. kick_requests(mdsc, mds);
  4210. }
  4211. __wake_requests(mdsc, &mdsc->waiting_for_map);
  4212. mutex_unlock(&mdsc->mutex);
  4213. }
  4214. static void ceph_mdsc_stop(struct ceph_mds_client *mdsc)
  4215. {
  4216. dout("stop\n");
  4217. /*
  4218. * Make sure the delayed work stopped before releasing
  4219. * the resources.
  4220. *
  4221. * Because the cancel_delayed_work_sync() will only
  4222. * guarantee that the work finishes executing. But the
  4223. * delayed work will re-arm itself again after that.
  4224. */
  4225. flush_delayed_work(&mdsc->delayed_work);
  4226. if (mdsc->mdsmap)
  4227. ceph_mdsmap_destroy(mdsc->mdsmap);
  4228. kfree(mdsc->sessions);
  4229. ceph_caps_finalize(mdsc);
  4230. ceph_pool_perm_destroy(mdsc);
  4231. }
  4232. void ceph_mdsc_destroy(struct ceph_fs_client *fsc)
  4233. {
  4234. struct ceph_mds_client *mdsc = fsc->mdsc;
  4235. dout("mdsc_destroy %p\n", mdsc);
  4236. if (!mdsc)
  4237. return;
  4238. /* flush out any connection work with references to us */
  4239. ceph_msgr_flush();
  4240. ceph_mdsc_stop(mdsc);
  4241. ceph_metric_destroy(&mdsc->metric);
  4242. fsc->mdsc = NULL;
  4243. kfree(mdsc);
  4244. dout("mdsc_destroy %p done\n", mdsc);
  4245. }
  4246. void ceph_mdsc_handle_fsmap(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
  4247. {
  4248. struct ceph_fs_client *fsc = mdsc->fsc;
  4249. const char *mds_namespace = fsc->mount_options->mds_namespace;
  4250. void *p = msg->front.iov_base;
  4251. void *end = p + msg->front.iov_len;
  4252. u32 epoch;
  4253. u32 map_len;
  4254. u32 num_fs;
  4255. u32 mount_fscid = (u32)-1;
  4256. u8 struct_v, struct_cv;
  4257. int err = -EINVAL;
  4258. ceph_decode_need(&p, end, sizeof(u32), bad);
  4259. epoch = ceph_decode_32(&p);
  4260. dout("handle_fsmap epoch %u\n", epoch);
  4261. ceph_decode_need(&p, end, 2 + sizeof(u32), bad);
  4262. struct_v = ceph_decode_8(&p);
  4263. struct_cv = ceph_decode_8(&p);
  4264. map_len = ceph_decode_32(&p);
  4265. ceph_decode_need(&p, end, sizeof(u32) * 3, bad);
  4266. p += sizeof(u32) * 2; /* skip epoch and legacy_client_fscid */
  4267. num_fs = ceph_decode_32(&p);
  4268. while (num_fs-- > 0) {
  4269. void *info_p, *info_end;
  4270. u32 info_len;
  4271. u8 info_v, info_cv;
  4272. u32 fscid, namelen;
  4273. ceph_decode_need(&p, end, 2 + sizeof(u32), bad);
  4274. info_v = ceph_decode_8(&p);
  4275. info_cv = ceph_decode_8(&p);
  4276. info_len = ceph_decode_32(&p);
  4277. ceph_decode_need(&p, end, info_len, bad);
  4278. info_p = p;
  4279. info_end = p + info_len;
  4280. p = info_end;
  4281. ceph_decode_need(&info_p, info_end, sizeof(u32) * 2, bad);
  4282. fscid = ceph_decode_32(&info_p);
  4283. namelen = ceph_decode_32(&info_p);
  4284. ceph_decode_need(&info_p, info_end, namelen, bad);
  4285. if (mds_namespace &&
  4286. strlen(mds_namespace) == namelen &&
  4287. !strncmp(mds_namespace, (char *)info_p, namelen)) {
  4288. mount_fscid = fscid;
  4289. break;
  4290. }
  4291. }
  4292. ceph_monc_got_map(&fsc->client->monc, CEPH_SUB_FSMAP, epoch);
  4293. if (mount_fscid != (u32)-1) {
  4294. fsc->client->monc.fs_cluster_id = mount_fscid;
  4295. ceph_monc_want_map(&fsc->client->monc, CEPH_SUB_MDSMAP,
  4296. 0, true);
  4297. ceph_monc_renew_subs(&fsc->client->monc);
  4298. } else {
  4299. err = -ENOENT;
  4300. goto err_out;
  4301. }
  4302. return;
  4303. bad:
  4304. pr_err("error decoding fsmap\n");
  4305. err_out:
  4306. mutex_lock(&mdsc->mutex);
  4307. mdsc->mdsmap_err = err;
  4308. __wake_requests(mdsc, &mdsc->waiting_for_map);
  4309. mutex_unlock(&mdsc->mutex);
  4310. }
  4311. /*
  4312. * handle mds map update.
  4313. */
  4314. void ceph_mdsc_handle_mdsmap(struct ceph_mds_client *mdsc, struct ceph_msg *msg)
  4315. {
  4316. u32 epoch;
  4317. u32 maplen;
  4318. void *p = msg->front.iov_base;
  4319. void *end = p + msg->front.iov_len;
  4320. struct ceph_mdsmap *newmap, *oldmap;
  4321. struct ceph_fsid fsid;
  4322. int err = -EINVAL;
  4323. ceph_decode_need(&p, end, sizeof(fsid)+2*sizeof(u32), bad);
  4324. ceph_decode_copy(&p, &fsid, sizeof(fsid));
  4325. if (ceph_check_fsid(mdsc->fsc->client, &fsid) < 0)
  4326. return;
  4327. epoch = ceph_decode_32(&p);
  4328. maplen = ceph_decode_32(&p);
  4329. dout("handle_map epoch %u len %d\n", epoch, (int)maplen);
  4330. /* do we need it? */
  4331. mutex_lock(&mdsc->mutex);
  4332. if (mdsc->mdsmap && epoch <= mdsc->mdsmap->m_epoch) {
  4333. dout("handle_map epoch %u <= our %u\n",
  4334. epoch, mdsc->mdsmap->m_epoch);
  4335. mutex_unlock(&mdsc->mutex);
  4336. return;
  4337. }
  4338. newmap = ceph_mdsmap_decode(&p, end);
  4339. if (IS_ERR(newmap)) {
  4340. err = PTR_ERR(newmap);
  4341. goto bad_unlock;
  4342. }
  4343. /* swap into place */
  4344. if (mdsc->mdsmap) {
  4345. oldmap = mdsc->mdsmap;
  4346. mdsc->mdsmap = newmap;
  4347. check_new_map(mdsc, newmap, oldmap);
  4348. ceph_mdsmap_destroy(oldmap);
  4349. } else {
  4350. mdsc->mdsmap = newmap; /* first mds map */
  4351. }
  4352. mdsc->fsc->max_file_size = min((loff_t)mdsc->mdsmap->m_max_file_size,
  4353. MAX_LFS_FILESIZE);
  4354. __wake_requests(mdsc, &mdsc->waiting_for_map);
  4355. ceph_monc_got_map(&mdsc->fsc->client->monc, CEPH_SUB_MDSMAP,
  4356. mdsc->mdsmap->m_epoch);
  4357. mutex_unlock(&mdsc->mutex);
  4358. schedule_delayed(mdsc, 0);
  4359. return;
  4360. bad_unlock:
  4361. mutex_unlock(&mdsc->mutex);
  4362. bad:
  4363. pr_err("error decoding mdsmap %d\n", err);
  4364. return;
  4365. }
  4366. static struct ceph_connection *con_get(struct ceph_connection *con)
  4367. {
  4368. struct ceph_mds_session *s = con->private;
  4369. if (ceph_get_mds_session(s))
  4370. return con;
  4371. return NULL;
  4372. }
  4373. static void con_put(struct ceph_connection *con)
  4374. {
  4375. struct ceph_mds_session *s = con->private;
  4376. ceph_put_mds_session(s);
  4377. }
  4378. /*
  4379. * if the client is unresponsive for long enough, the mds will kill
  4380. * the session entirely.
  4381. */
  4382. static void peer_reset(struct ceph_connection *con)
  4383. {
  4384. struct ceph_mds_session *s = con->private;
  4385. struct ceph_mds_client *mdsc = s->s_mdsc;
  4386. pr_warn("mds%d closed our session\n", s->s_mds);
  4387. send_mds_reconnect(mdsc, s);
  4388. }
  4389. static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
  4390. {
  4391. struct ceph_mds_session *s = con->private;
  4392. struct ceph_mds_client *mdsc = s->s_mdsc;
  4393. int type = le16_to_cpu(msg->hdr.type);
  4394. mutex_lock(&mdsc->mutex);
  4395. if (__verify_registered_session(mdsc, s) < 0) {
  4396. mutex_unlock(&mdsc->mutex);
  4397. goto out;
  4398. }
  4399. mutex_unlock(&mdsc->mutex);
  4400. switch (type) {
  4401. case CEPH_MSG_MDS_MAP:
  4402. ceph_mdsc_handle_mdsmap(mdsc, msg);
  4403. break;
  4404. case CEPH_MSG_FS_MAP_USER:
  4405. ceph_mdsc_handle_fsmap(mdsc, msg);
  4406. break;
  4407. case CEPH_MSG_CLIENT_SESSION:
  4408. handle_session(s, msg);
  4409. break;
  4410. case CEPH_MSG_CLIENT_REPLY:
  4411. handle_reply(s, msg);
  4412. break;
  4413. case CEPH_MSG_CLIENT_REQUEST_FORWARD:
  4414. handle_forward(mdsc, s, msg);
  4415. break;
  4416. case CEPH_MSG_CLIENT_CAPS:
  4417. ceph_handle_caps(s, msg);
  4418. break;
  4419. case CEPH_MSG_CLIENT_SNAP:
  4420. ceph_handle_snap(mdsc, s, msg);
  4421. break;
  4422. case CEPH_MSG_CLIENT_LEASE:
  4423. handle_lease(mdsc, s, msg);
  4424. break;
  4425. case CEPH_MSG_CLIENT_QUOTA:
  4426. ceph_handle_quota(mdsc, s, msg);
  4427. break;
  4428. default:
  4429. pr_err("received unknown message type %d %s\n", type,
  4430. ceph_msg_type_name(type));
  4431. }
  4432. out:
  4433. ceph_msg_put(msg);
  4434. }
  4435. /*
  4436. * authentication
  4437. */
  4438. /*
  4439. * Note: returned pointer is the address of a structure that's
  4440. * managed separately. Caller must *not* attempt to free it.
  4441. */
  4442. static struct ceph_auth_handshake *get_authorizer(struct ceph_connection *con,
  4443. int *proto, int force_new)
  4444. {
  4445. struct ceph_mds_session *s = con->private;
  4446. struct ceph_mds_client *mdsc = s->s_mdsc;
  4447. struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
  4448. struct ceph_auth_handshake *auth = &s->s_auth;
  4449. if (force_new && auth->authorizer) {
  4450. ceph_auth_destroy_authorizer(auth->authorizer);
  4451. auth->authorizer = NULL;
  4452. }
  4453. if (!auth->authorizer) {
  4454. int ret = ceph_auth_create_authorizer(ac, CEPH_ENTITY_TYPE_MDS,
  4455. auth);
  4456. if (ret)
  4457. return ERR_PTR(ret);
  4458. } else {
  4459. int ret = ceph_auth_update_authorizer(ac, CEPH_ENTITY_TYPE_MDS,
  4460. auth);
  4461. if (ret)
  4462. return ERR_PTR(ret);
  4463. }
  4464. *proto = ac->protocol;
  4465. return auth;
  4466. }
  4467. static int add_authorizer_challenge(struct ceph_connection *con,
  4468. void *challenge_buf, int challenge_buf_len)
  4469. {
  4470. struct ceph_mds_session *s = con->private;
  4471. struct ceph_mds_client *mdsc = s->s_mdsc;
  4472. struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
  4473. return ceph_auth_add_authorizer_challenge(ac, s->s_auth.authorizer,
  4474. challenge_buf, challenge_buf_len);
  4475. }
  4476. static int verify_authorizer_reply(struct ceph_connection *con)
  4477. {
  4478. struct ceph_mds_session *s = con->private;
  4479. struct ceph_mds_client *mdsc = s->s_mdsc;
  4480. struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
  4481. return ceph_auth_verify_authorizer_reply(ac, s->s_auth.authorizer);
  4482. }
  4483. static int invalidate_authorizer(struct ceph_connection *con)
  4484. {
  4485. struct ceph_mds_session *s = con->private;
  4486. struct ceph_mds_client *mdsc = s->s_mdsc;
  4487. struct ceph_auth_client *ac = mdsc->fsc->client->monc.auth;
  4488. ceph_auth_invalidate_authorizer(ac, CEPH_ENTITY_TYPE_MDS);
  4489. return ceph_monc_validate_auth(&mdsc->fsc->client->monc);
  4490. }
  4491. static struct ceph_msg *mds_alloc_msg(struct ceph_connection *con,
  4492. struct ceph_msg_header *hdr, int *skip)
  4493. {
  4494. struct ceph_msg *msg;
  4495. int type = (int) le16_to_cpu(hdr->type);
  4496. int front_len = (int) le32_to_cpu(hdr->front_len);
  4497. if (con->in_msg)
  4498. return con->in_msg;
  4499. *skip = 0;
  4500. msg = ceph_msg_new(type, front_len, GFP_NOFS, false);
  4501. if (!msg) {
  4502. pr_err("unable to allocate msg type %d len %d\n",
  4503. type, front_len);
  4504. return NULL;
  4505. }
  4506. return msg;
  4507. }
  4508. static int mds_sign_message(struct ceph_msg *msg)
  4509. {
  4510. struct ceph_mds_session *s = msg->con->private;
  4511. struct ceph_auth_handshake *auth = &s->s_auth;
  4512. return ceph_auth_sign_message(auth, msg);
  4513. }
  4514. static int mds_check_message_signature(struct ceph_msg *msg)
  4515. {
  4516. struct ceph_mds_session *s = msg->con->private;
  4517. struct ceph_auth_handshake *auth = &s->s_auth;
  4518. return ceph_auth_check_message_signature(auth, msg);
  4519. }
  4520. static const struct ceph_connection_operations mds_con_ops = {
  4521. .get = con_get,
  4522. .put = con_put,
  4523. .dispatch = dispatch,
  4524. .get_authorizer = get_authorizer,
  4525. .add_authorizer_challenge = add_authorizer_challenge,
  4526. .verify_authorizer_reply = verify_authorizer_reply,
  4527. .invalidate_authorizer = invalidate_authorizer,
  4528. .peer_reset = peer_reset,
  4529. .alloc_msg = mds_alloc_msg,
  4530. .sign_message = mds_sign_message,
  4531. .check_message_signature = mds_check_message_signature,
  4532. };
  4533. /* eof */