ceph/src/msg/Message.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #ifdef ENCODE_DUMP
   5 # include <typeinfo>
   6 # include <cxxabi.h>
   7 #endif
   8
   9 #include <iostream>
  10 using namespace std;
  11
  12 #include "include/types.h"
  13
  14 #include "global/global_context.h"
  15
  16 #include "Message.h"
  17
  18 #include "messages/MPGStats.h"
  19
  20 #include "messages/MGenericMessage.h"
  21
  22 #include "messages/MPGStatsAck.h"
  23
  24 #include "messages/MStatfs.h"
  25 #include "messages/MStatfsReply.h"
  26
  27 #include "messages/MGetPoolStats.h"
  28 #include "messages/MGetPoolStatsReply.h"
  29
  30
  31 #include "messages/MPoolOp.h"
  32 #include "messages/MPoolOpReply.h"
  33
  34 #include "messages/PaxosServiceMessage.h"
  35 #include "messages/MMonCommand.h"
  36 #include "messages/MMonCommandAck.h"
  37 #include "messages/MMonPaxos.h"
  38
  39 #include "messages/MMonProbe.h"
  40 #include "messages/MMonJoin.h"
  41 #include "messages/MMonElection.h"
  42 #include "messages/MMonSync.h"
  43 #include "messages/MMonScrub.h"
  44
  45 #include "messages/MLog.h"
  46 #include "messages/MLogAck.h"
  47
  48 #include "messages/MPing.h"
  49
  50 #include "messages/MCommand.h"
  51 #include "messages/MCommandReply.h"
  52 #include "messages/MBackfillReserve.h"
  53 #include "messages/MRecoveryReserve.h"
  54
  55 #include "messages/MRoute.h"
  56 #include "messages/MForward.h"
  57
  58 #include "messages/MOSDBoot.h"
  59 #include "messages/MOSDAlive.h"
  60 #include "messages/MOSDBeacon.h"
  61 #include "messages/MOSDPGTemp.h"
  62 #include "messages/MOSDFailure.h"
  63 #include "messages/MOSDMarkMeDown.h"
  64 #include "messages/MOSDFull.h"
  65 #include "messages/MOSDPing.h"
  66 #include "messages/MOSDOp.h"
  67 #include "messages/MOSDOpReply.h"
  68 #include "messages/MOSDSubOp.h"
  69 #include "messages/MOSDSubOpReply.h"
  70 #include "messages/MOSDRepOp.h"
  71 #include "messages/MOSDRepOpReply.h"
  72 #include "messages/MOSDMap.h"
  73 #include "messages/MMonGetOSDMap.h"
  74
  75 #include "messages/MOSDPGCreated.h"
  76 #include "messages/MOSDPGNotify.h"
  77 #include "messages/MOSDPGQuery.h"
  78 #include "messages/MOSDPGLog.h"
  79 #include "messages/MOSDPGRemove.h"
  80 #include "messages/MOSDPGInfo.h"
  81 #include "messages/MOSDPGCreate.h"
  82 #include "messages/MOSDPGTrim.h"
  83 #include "messages/MOSDScrub.h"
  84 #include "messages/MOSDScrubReserve.h"
  85 #include "messages/MOSDRepScrub.h"
  86 #include "messages/MOSDRepScrubMap.h"
  87 #include "messages/MOSDPGScan.h"
  88 #include "messages/MOSDPGBackfill.h"
  89 #include "messages/MOSDBackoff.h"
  90 #include "messages/MOSDPGBackfillRemove.h"
  91
  92 #include "messages/MRemoveSnaps.h"
  93
  94 #include "messages/MMonMap.h"
  95 #include "messages/MMonGetMap.h"
  96 #include "messages/MMonGetVersion.h"
  97 #include "messages/MMonGetVersionReply.h"
  98 #include "messages/MMonHealth.h"
  99 #include "messages/MMonMetadata.h"
 100 #include "messages/MDataPing.h"
 101 #include "messages/MAuth.h"
 102 #include "messages/MAuthReply.h"
 103 #include "messages/MMonSubscribe.h"
 104 #include "messages/MMonSubscribeAck.h"
 105 #include "messages/MMonGlobalID.h"
 106 #include "messages/MClientSession.h"
 107 #include "messages/MClientReconnect.h"
 108 #include "messages/MClientRequest.h"
 109 #include "messages/MClientRequestForward.h"
 110 #include "messages/MClientReply.h"
 111 #include "messages/MClientCaps.h"
 112 #include "messages/MClientCapRelease.h"
 113 #include "messages/MClientLease.h"
 114 #include "messages/MClientSnap.h"
 115 #include "messages/MClientQuota.h"
 116
 117 #include "messages/MMDSSlaveRequest.h"
 118
 119 #include "messages/MMDSMap.h"
 120 #include "messages/MFSMap.h"
 121 #include "messages/MFSMapUser.h"
 122 #include "messages/MMDSBeacon.h"
 123 #include "messages/MMDSLoadTargets.h"
 124 #include "messages/MMDSResolve.h"
 125 #include "messages/MMDSResolveAck.h"
 126 #include "messages/MMDSCacheRejoin.h"
 127 #include "messages/MMDSFindIno.h"
 128 #include "messages/MMDSFindInoReply.h"
 129 #include "messages/MMDSOpenIno.h"
 130 #include "messages/MMDSOpenInoReply.h"
 131
 132 #include "messages/MDirUpdate.h"
 133 #include "messages/MDiscover.h"
 134 #include "messages/MDiscoverReply.h"
 135
 136 #include "messages/MMDSFragmentNotify.h"
 137
 138 #include "messages/MExportDirDiscover.h"
 139 #include "messages/MExportDirDiscoverAck.h"
 140 #include "messages/MExportDirCancel.h"
 141 #include "messages/MExportDirPrep.h"
 142 #include "messages/MExportDirPrepAck.h"
 143 #include "messages/MExportDir.h"
 144 #include "messages/MExportDirAck.h"
 145 #include "messages/MExportDirNotify.h"
 146 #include "messages/MExportDirNotifyAck.h"
 147 #include "messages/MExportDirFinish.h"
 148
 149 #include "messages/MExportCaps.h"
 150 #include "messages/MExportCapsAck.h"
 151 #include "messages/MGatherCaps.h"
 152
 153
 154 #include "messages/MDentryUnlink.h"
 155 #include "messages/MDentryLink.h"
 156
 157 #include "messages/MHeartbeat.h"
 158
 159 #include "messages/MMDSTableRequest.h"
 160
 161 //#include "messages/MInodeUpdate.h"
 162 #include "messages/MCacheExpire.h"
 163 #include "messages/MInodeFileCaps.h"
 164
 165 #include "messages/MMgrBeacon.h"
 166 #include "messages/MMgrMap.h"
 167 #include "messages/MMgrDigest.h"
 168 #include "messages/MMgrReport.h"
 169 #include "messages/MMgrOpen.h"
 170 #include "messages/MMgrConfigure.h"
 171 #include "messages/MMonMgrReport.h"
 172
 173 #include "messages/MLock.h"
 174
 175 #include "messages/MWatchNotify.h"
 176 #include "messages/MTimeCheck.h"
 177
 178 #include "common/config.h"
 179
 180 #include "messages/MOSDPGPush.h"
 181 #include "messages/MOSDPGPushReply.h"
 182 #include "messages/MOSDPGPull.h"
 183
 184 #include "messages/MOSDECSubOpWrite.h"
 185 #include "messages/MOSDECSubOpWriteReply.h"
 186 #include "messages/MOSDECSubOpRead.h"
 187 #include "messages/MOSDECSubOpReadReply.h"
 188
 189 #include "messages/MOSDPGUpdateLogMissing.h"
 190 #include "messages/MOSDPGUpdateLogMissingReply.h"
 191
 192 #define DEBUGLVL  10    // debug level of output
 193
 194 #define dout_subsys ceph_subsys_ms
 195
 196 void Message::encode(uint64_t features, int crcflags)
 197 {
 198   // encode and copy out of *m
 199   if (empty_payload()) {
 200     assert(middle.length() == 0);
 201     encode_payload(features);
 202
 203     if (byte_throttler) {
 204       byte_throttler->take(payload.length() + middle.length());
 205     }
 206
 207     // if the encoder didn't specify past compatibility, we assume it
 208     // is incompatible.
 209     if (header.compat_version == 0)
 210       header.compat_version = header.version;
 211   }
 212   if (crcflags & MSG_CRC_HEADER)
 213     calc_front_crc();
 214
 215   // update envelope
 216   header.front_len = get_payload().length();
 217   header.middle_len = get_middle().length();
 218   header.data_len = get_data().length();
 219   if (crcflags & MSG_CRC_HEADER)
 220     calc_header_crc();
 221
 222   footer.flags = CEPH_MSG_FOOTER_COMPLETE;
 223
 224   if (crcflags & MSG_CRC_DATA) {
 225     calc_data_crc();
 226
 227 #ifdef ENCODE_DUMP
 228     bufferlist bl;
 229     ::encode(get_header(), bl);
 230
 231     // dump the old footer format
 232     ceph_msg_footer_old old_footer;
 233     old_footer.front_crc = footer.front_crc;
 234     old_footer.middle_crc = footer.middle_crc;
 235     old_footer.data_crc = footer.data_crc;
 236     old_footer.flags = footer.flags;
 237     ::encode(old_footer, bl);
 238
 239     ::encode(get_payload(), bl);
 240     ::encode(get_middle(), bl);
 241     ::encode(get_data(), bl);
 242
 243     // this is almost an exponential backoff, except because we count
 244     // bits we tend to sample things we encode later, which should be
 245     // more representative.
 246     static int i = 0;
 247     i++;
 248     int bits = 0;
 249     for (unsigned t = i; t; bits++)
 250       t &= t - 1;
 251     if (bits <= 2) {
 252       char fn[200];
 253       int status;
 254       snprintf(fn, sizeof(fn), ENCODE_STRINGIFY(ENCODE_DUMP) "/%s__%d.%x",
 255                abi::__cxa_demangle(typeid(*this).name(), 0, 0, &status),
 256                getpid(), i++);
 257       int fd = ::open(fn, O_WRONLY|O_TRUNC|O_CREAT, 0644);
 258       if (fd >= 0) {
 259         bl.write_fd(fd);
 260         ::close(fd);
 261       }
 262     }
 263 #endif
 264   } else {
 265     footer.flags = (unsigned)footer.flags | CEPH_MSG_FOOTER_NOCRC;
 266   }
 267 }
 268
 269 void Message::dump(Formatter *f) const
 270 {
 271   stringstream ss;
 272   print(ss);
 273   f->dump_string("summary", ss.str());
 274 }
 275
 276 Message *decode_message(CephContext *cct, int crcflags,
 277                         ceph_msg_header& header,
 278                         ceph_msg_footer& footer,
 279                         bufferlist& front, bufferlist& middle,
 280                         bufferlist& data, Connection* conn)
 281 {
 282   // verify crc
 283   if (crcflags & MSG_CRC_HEADER) {
 284     __u32 front_crc = front.crc32c(0);
 285     __u32 middle_crc = middle.crc32c(0);
 286
 287     if (front_crc != footer.front_crc) {
 288       if (cct) {
 289         ldout(cct, 0) << "bad crc in front " << front_crc << " != exp " << footer.front_crc << dendl;
 290         ldout(cct, 20) << " ";
 291         front.hexdump(*_dout);
 292         *_dout << dendl;
 293       }
 294       return 0;
 295     }
 296     if (middle_crc != footer.middle_crc) {
 297       if (cct) {
 298         ldout(cct, 0) << "bad crc in middle " << middle_crc << " != exp " << footer.middle_crc << dendl;
 299         ldout(cct, 20) << " ";
 300         middle.hexdump(*_dout);
 301         *_dout << dendl;
 302       }
 303       return 0;
 304     }
 305   }
 306   if (crcflags & MSG_CRC_DATA) {
 307     if ((footer.flags & CEPH_MSG_FOOTER_NOCRC) == 0) {
 308       __u32 data_crc = data.crc32c(0);
 309       if (data_crc != footer.data_crc) {
 310         if (cct) {
 311           ldout(cct, 0) << "bad crc in data " << data_crc << " != exp " << footer.data_crc << dendl;
 312           ldout(cct, 20) << " ";
 313           data.hexdump(*_dout);
 314           *_dout << dendl;
 315         }
 316         return 0;
 317       }
 318     }
 319   }
 320
 321   // make message
 322   Message *m = 0;
 323   int type = header.type;
 324   switch (type) {
 325
 326     // -- with payload --
 327
 328   case MSG_PGSTATS:
 329     m = new MPGStats;
 330     break;
 331   case MSG_PGSTATSACK:
 332     m = new MPGStatsAck;
 333     break;
 334
 335   case CEPH_MSG_STATFS:
 336     m = new MStatfs;
 337     break;
 338   case CEPH_MSG_STATFS_REPLY:
 339     m = new MStatfsReply;
 340     break;
 341   case MSG_GETPOOLSTATS:
 342     m = new MGetPoolStats;
 343     break;
 344   case MSG_GETPOOLSTATSREPLY:
 345     m = new MGetPoolStatsReply;
 346     break;
 347   case CEPH_MSG_POOLOP:
 348     m = new MPoolOp;
 349     break;
 350   case CEPH_MSG_POOLOP_REPLY:
 351     m = new MPoolOpReply;
 352     break;
 353   case MSG_MON_COMMAND:
 354     m = new MMonCommand;
 355     break;
 356   case MSG_MON_COMMAND_ACK:
 357     m = new MMonCommandAck;
 358     break;
 359   case MSG_MON_PAXOS:
 360     m = new MMonPaxos;
 361     break;
 362
 363   case MSG_MON_PROBE:
 364     m = new MMonProbe;
 365     break;
 366   case MSG_MON_JOIN:
 367     m = new MMonJoin;
 368     break;
 369   case MSG_MON_ELECTION:
 370     m = new MMonElection;
 371     break;
 372   case MSG_MON_SYNC:
 373     m = new MMonSync;
 374     break;
 375   case MSG_MON_SCRUB:
 376     m = new MMonScrub;
 377     break;
 378
 379   case MSG_LOG:
 380     m = new MLog;
 381     break;
 382   case MSG_LOGACK:
 383     m = new MLogAck;
 384     break;
 385
 386   case CEPH_MSG_PING:
 387     m = new MPing();
 388     break;
 389   case MSG_COMMAND:
 390     m = new MCommand;
 391     break;
 392   case MSG_COMMAND_REPLY:
 393     m = new MCommandReply;
 394     break;
 395   case MSG_OSD_BACKFILL_RESERVE:
 396     m = new MBackfillReserve;
 397     break;
 398   case MSG_OSD_RECOVERY_RESERVE:
 399     m = new MRecoveryReserve;
 400     break;
 401
 402   case MSG_ROUTE:
 403     m = new MRoute;
 404     break;
 405   case MSG_FORWARD:
 406     m = new MForward;
 407     break;
 408
 409   case CEPH_MSG_MON_MAP:
 410     m = new MMonMap;
 411     break;
 412   case CEPH_MSG_MON_GET_MAP:
 413     m = new MMonGetMap;
 414     break;
 415   case CEPH_MSG_MON_GET_OSDMAP:
 416     m = new MMonGetOSDMap;
 417     break;
 418   case CEPH_MSG_MON_GET_VERSION:
 419     m = new MMonGetVersion();
 420     break;
 421   case CEPH_MSG_MON_GET_VERSION_REPLY:
 422     m = new MMonGetVersionReply();
 423     break;
 424   case CEPH_MSG_MON_METADATA:
 425     m = new MMonMetadata();
 426     break;
 427
 428   case MSG_OSD_BOOT:
 429     m = new MOSDBoot();
 430     break;
 431   case MSG_OSD_ALIVE:
 432     m = new MOSDAlive();
 433     break;
 434   case MSG_OSD_BEACON:
 435     m = new MOSDBeacon();
 436     break;
 437   case MSG_OSD_PGTEMP:
 438     m = new MOSDPGTemp;
 439     break;
 440   case MSG_OSD_FAILURE:
 441     m = new MOSDFailure();
 442     break;
 443   case MSG_OSD_MARK_ME_DOWN:
 444     m = new MOSDMarkMeDown();
 445     break;
 446   case MSG_OSD_FULL:
 447     m = new MOSDFull();
 448     break;
 449   case MSG_OSD_PING:
 450     m = new MOSDPing();
 451     break;
 452   case CEPH_MSG_OSD_OP:
 453     m = new MOSDOp();
 454     break;
 455   case CEPH_MSG_OSD_OPREPLY:
 456     m = new MOSDOpReply();
 457     break;
 458   case MSG_OSD_SUBOP:
 459     m = new MOSDSubOp();
 460     break;
 461   case MSG_OSD_SUBOPREPLY:
 462     m = new MOSDSubOpReply();
 463     break;
 464   case MSG_OSD_REPOP:
 465     m = new MOSDRepOp();
 466     break;
 467   case MSG_OSD_REPOPREPLY:
 468     m = new MOSDRepOpReply();
 469     break;
 470   case MSG_OSD_PG_CREATED:
 471     m = new MOSDPGCreated();
 472     break;
 473   case MSG_OSD_PG_UPDATE_LOG_MISSING:
 474     m = new MOSDPGUpdateLogMissing();
 475     break;
 476   case MSG_OSD_PG_UPDATE_LOG_MISSING_REPLY:
 477     m = new MOSDPGUpdateLogMissingReply();
 478     break;
 479   case CEPH_MSG_OSD_BACKOFF:
 480     m = new MOSDBackoff;
 481     break;
 482
 483   case CEPH_MSG_OSD_MAP:
 484     m = new MOSDMap;
 485     break;
 486
 487   case CEPH_MSG_WATCH_NOTIFY:
 488     m = new MWatchNotify;
 489     break;
 490
 491   case MSG_OSD_PG_NOTIFY:
 492     m = new MOSDPGNotify;
 493     break;
 494   case MSG_OSD_PG_QUERY:
 495     m = new MOSDPGQuery;
 496     break;
 497   case MSG_OSD_PG_LOG:
 498     m = new MOSDPGLog;
 499     break;
 500   case MSG_OSD_PG_REMOVE:
 501     m = new MOSDPGRemove;
 502     break;
 503   case MSG_OSD_PG_INFO:
 504     m = new MOSDPGInfo;
 505     break;
 506   case MSG_OSD_PG_CREATE:
 507     m = new MOSDPGCreate;
 508     break;
 509   case MSG_OSD_PG_TRIM:
 510     m = new MOSDPGTrim;
 511     break;
 512
 513   case MSG_OSD_SCRUB:
 514     m = new MOSDScrub;
 515     break;
 516   case MSG_OSD_SCRUB_RESERVE:
 517     m = new MOSDScrubReserve;
 518     break;
 519   case MSG_REMOVE_SNAPS:
 520     m = new MRemoveSnaps;
 521     break;
 522   case MSG_OSD_REP_SCRUB:
 523     m = new MOSDRepScrub;
 524     break;
 525   case MSG_OSD_REP_SCRUBMAP:
 526     m = new MOSDRepScrubMap;
 527     break;
 528   case MSG_OSD_PG_SCAN:
 529     m = new MOSDPGScan;
 530     break;
 531   case MSG_OSD_PG_BACKFILL:
 532     m = new MOSDPGBackfill;
 533     break;
 534   case MSG_OSD_PG_BACKFILL_REMOVE:
 535     m = new MOSDPGBackfillRemove;
 536     break;
 537   case MSG_OSD_PG_PUSH:
 538     m = new MOSDPGPush;
 539     break;
 540   case MSG_OSD_PG_PULL:
 541     m = new MOSDPGPull;
 542     break;
 543   case MSG_OSD_PG_PUSH_REPLY:
 544     m = new MOSDPGPushReply;
 545     break;
 546   case MSG_OSD_EC_WRITE:
 547     m = new MOSDECSubOpWrite;
 548     break;
 549   case MSG_OSD_EC_WRITE_REPLY:
 550     m = new MOSDECSubOpWriteReply;
 551     break;
 552   case MSG_OSD_EC_READ:
 553     m = new MOSDECSubOpRead;
 554     break;
 555   case MSG_OSD_EC_READ_REPLY:
 556     m = new MOSDECSubOpReadReply;
 557     break;
 558    // auth
 559   case CEPH_MSG_AUTH:
 560     m = new MAuth;
 561     break;
 562   case CEPH_MSG_AUTH_REPLY:
 563     m = new MAuthReply;
 564     break;
 565
 566   case MSG_MON_GLOBAL_ID:
 567     m = new MMonGlobalID;
 568     break;
 569
 570     // clients
 571   case CEPH_MSG_MON_SUBSCRIBE:
 572     m = new MMonSubscribe;
 573     break;
 574   case CEPH_MSG_MON_SUBSCRIBE_ACK:
 575     m = new MMonSubscribeAck;
 576     break;
 577   case CEPH_MSG_CLIENT_SESSION:
 578     m = new MClientSession;
 579     break;
 580   case CEPH_MSG_CLIENT_RECONNECT:
 581     m = new MClientReconnect;
 582     break;
 583   case CEPH_MSG_CLIENT_REQUEST:
 584     m = new MClientRequest;
 585     break;
 586   case CEPH_MSG_CLIENT_REQUEST_FORWARD:
 587     m = new MClientRequestForward;
 588     break;
 589   case CEPH_MSG_CLIENT_REPLY:
 590     m = new MClientReply;
 591     break;
 592   case CEPH_MSG_CLIENT_CAPS:
 593     m = new MClientCaps;
 594     break;
 595   case CEPH_MSG_CLIENT_CAPRELEASE:
 596     m = new MClientCapRelease;
 597     break;
 598   case CEPH_MSG_CLIENT_LEASE:
 599     m = new MClientLease;
 600     break;
 601   case CEPH_MSG_CLIENT_SNAP:
 602     m = new MClientSnap;
 603     break;
 604   case CEPH_MSG_CLIENT_QUOTA:
 605     m = new MClientQuota;
 606     break;
 607
 608     // mds
 609   case MSG_MDS_SLAVE_REQUEST:
 610     m = new MMDSSlaveRequest;
 611     break;
 612
 613   case CEPH_MSG_MDS_MAP:
 614     m = new MMDSMap;
 615     break;
 616   case CEPH_MSG_FS_MAP:
 617     m = new MFSMap;
 618     break;
 619   case CEPH_MSG_FS_MAP_USER:
 620     m = new MFSMapUser;
 621     break;
 622   case MSG_MDS_BEACON:
 623     m = new MMDSBeacon;
 624     break;
 625   case MSG_MDS_OFFLOAD_TARGETS:
 626     m = new MMDSLoadTargets;
 627     break;
 628   case MSG_MDS_RESOLVE:
 629     m = new MMDSResolve;
 630     break;
 631   case MSG_MDS_RESOLVEACK:
 632     m = new MMDSResolveAck;
 633     break;
 634   case MSG_MDS_CACHEREJOIN:
 635     m = new MMDSCacheRejoin;
 636         break;
 637
 638   case MSG_MDS_DIRUPDATE:
 639     m = new MDirUpdate();
 640     break;
 641
 642   case MSG_MDS_DISCOVER:
 643     m = new MDiscover();
 644     break;
 645   case MSG_MDS_DISCOVERREPLY:
 646     m = new MDiscoverReply();
 647     break;
 648
 649   case MSG_MDS_FINDINO:
 650     m = new MMDSFindIno;
 651     break;
 652   case MSG_MDS_FINDINOREPLY:
 653     m = new MMDSFindInoReply;
 654     break;
 655
 656   case MSG_MDS_OPENINO:
 657     m = new MMDSOpenIno;
 658     break;
 659   case MSG_MDS_OPENINOREPLY:
 660     m = new MMDSOpenInoReply;
 661     break;
 662
 663   case MSG_MDS_FRAGMENTNOTIFY:
 664     m = new MMDSFragmentNotify;
 665     break;
 666
 667   case MSG_MDS_EXPORTDIRDISCOVER:
 668     m = new MExportDirDiscover();
 669     break;
 670   case MSG_MDS_EXPORTDIRDISCOVERACK:
 671     m = new MExportDirDiscoverAck();
 672     break;
 673   case MSG_MDS_EXPORTDIRCANCEL:
 674     m = new MExportDirCancel();
 675     break;
 676
 677   case MSG_MDS_EXPORTDIR:
 678     m = new MExportDir;
 679     break;
 680   case MSG_MDS_EXPORTDIRACK:
 681     m = new MExportDirAck;
 682     break;
 683   case MSG_MDS_EXPORTDIRFINISH:
 684     m = new MExportDirFinish;
 685     break;
 686
 687   case MSG_MDS_EXPORTDIRNOTIFY:
 688     m = new MExportDirNotify();
 689     break;
 690
 691   case MSG_MDS_EXPORTDIRNOTIFYACK:
 692     m = new MExportDirNotifyAck();
 693     break;
 694
 695   case MSG_MDS_EXPORTDIRPREP:
 696     m = new MExportDirPrep();
 697     break;
 698
 699   case MSG_MDS_EXPORTDIRPREPACK:
 700     m = new MExportDirPrepAck();
 701     break;
 702
 703   case MSG_MDS_EXPORTCAPS:
 704     m = new MExportCaps;
 705     break;
 706   case MSG_MDS_EXPORTCAPSACK:
 707     m = new MExportCapsAck;
 708     break;
 709   case MSG_MDS_GATHERCAPS:
 710     m = new MGatherCaps;
 711     break;
 712
 713
 714   case MSG_MDS_DENTRYUNLINK:
 715     m = new MDentryUnlink;
 716     break;
 717   case MSG_MDS_DENTRYLINK:
 718     m = new MDentryLink;
 719     break;
 720
 721   case MSG_MDS_HEARTBEAT:
 722     m = new MHeartbeat();
 723     break;
 724
 725   case MSG_MDS_CACHEEXPIRE:
 726     m = new MCacheExpire();
 727     break;
 728
 729   case MSG_MDS_TABLE_REQUEST:
 730     m = new MMDSTableRequest;
 731     break;
 732
 733         /*  case MSG_MDS_INODEUPDATE:
 734     m = new MInodeUpdate();
 735     break;
 736         */
 737
 738   case MSG_MDS_INODEFILECAPS:
 739     m = new MInodeFileCaps();
 740     break;
 741
 742   case MSG_MDS_LOCK:
 743     m = new MLock();
 744     break;
 745
 746   case MSG_MGR_BEACON:
 747     m = new MMgrBeacon();
 748     break;
 749
 750   case MSG_MON_MGR_REPORT:
 751     m = new MMonMgrReport();
 752     break;
 753
 754   case MSG_MGR_MAP:
 755     m = new MMgrMap();
 756     break;
 757
 758   case MSG_MGR_DIGEST:
 759     m = new MMgrDigest();
 760     break;
 761
 762   case MSG_MGR_OPEN:
 763     m = new MMgrOpen();
 764     break;
 765
 766   case MSG_MGR_REPORT:
 767     m = new MMgrReport();
 768     break;
 769
 770   case MSG_MGR_CONFIGURE:
 771     m = new MMgrConfigure();
 772     break;
 773
 774   case MSG_TIMECHECK:
 775     m = new MTimeCheck();
 776     break;
 777
 778   case MSG_MON_HEALTH:
 779     m = new MMonHealth();
 780     break;
 781 #if defined(HAVE_XIO)
 782   case MSG_DATA_PING:
 783     m = new MDataPing();
 784     break;
 785 #endif
 786     // -- simple messages without payload --
 787
 788   case CEPH_MSG_SHUTDOWN:
 789     m = new MGenericMessage(type);
 790     break;
 791
 792   default:
 793     if (cct) {
 794       ldout(cct, 0) << "can't decode unknown message type " << type << " MSG_AUTH=" << CEPH_MSG_AUTH << dendl;
 795       if (cct->_conf->ms_die_on_bad_msg)
 796         ceph_abort();
 797     }
 798     return 0;
 799   }
 800
 801   m->set_cct(cct);
 802
 803   // m->header.version, if non-zero, should be populated with the
 804   // newest version of the encoding the code supports.  If set, check
 805   // it against compat_version.
 806   if (m->get_header().version &&
 807       m->get_header().version < header.compat_version) {
 808     if (cct) {
 809       ldout(cct, 0) << "will not decode message of type " << type
 810                     << " version " << header.version
 811                     << " because compat_version " << header.compat_version
 812                     << " > supported version " << m->get_header().version << dendl;
 813       if (cct->_conf->ms_die_on_bad_msg)
 814         ceph_abort();
 815     }
 816     m->put();
 817     return 0;
 818   }
 819
 820   m->set_connection(conn);
 821   m->set_header(header);
 822   m->set_footer(footer);
 823   m->set_payload(front);
 824   m->set_middle(middle);
 825   m->set_data(data);
 826
 827   try {
 828     m->decode_payload();
 829   }
 830   catch (const buffer::error &e) {
 831     if (cct) {
 832       lderr(cct) << "failed to decode message of type " << type
 833                  << " v" << header.version
 834                  << ": " << e.what() << dendl;
 835       ldout(cct, cct->_conf->ms_dump_corrupt_message_level) << "dump: \n";
 836       m->get_payload().hexdump(*_dout);
 837       *_dout << dendl;
 838       if (cct->_conf->ms_die_on_bad_msg)
 839         ceph_abort();
 840     }
 841     m->put();
 842     return 0;
 843   }
 844
 845   // done!
 846   return m;
 847 }
 848
 849 void Message::encode_trace(bufferlist &bl, uint64_t features) const
 850 {
 851   auto p = trace.get_info();
 852   static const blkin_trace_info empty = { 0, 0, 0 };
 853   if (!p) {
 854     p = &empty;
 855   }
 856   ::encode(*p, bl);
 857 }
 858
 859 void Message::decode_trace(bufferlist::iterator &p, bool create)
 860 {
 861   blkin_trace_info info = {};
 862   ::decode(info, p);
 863
 864 #ifdef WITH_BLKIN
 865   if (!connection)
 866     return;
 867
 868   const auto msgr = connection->get_messenger();
 869   const auto endpoint = msgr->get_trace_endpoint();
 870   if (info.trace_id) {
 871     trace.init(get_type_name(), endpoint, &info, true);
 872     trace.event("decoded trace");
 873   } else if (create || (msgr->get_myname().is_osd() &&
 874                         msgr->cct->_conf->osd_blkin_trace_all)) {
 875     // create a trace even if we didn't get one on the wire
 876     trace.init(get_type_name(), endpoint);
 877     trace.event("created trace");
 878   }
 879   trace.keyval("tid", get_tid());
 880   trace.keyval("entity type", get_source().type_str());
 881   trace.keyval("entity num", get_source().num());
 882 #endif
 883 }
 884
 885
 886 // This routine is not used for ordinary messages, but only when encapsulating a message
 887 // for forwarding and routing.  It's also used in a backward compatibility test, which only
 888 // effectively tests backward compability for those functions.  To avoid backward compatibility
 889 // problems, we currently always encode and decode using the old footer format that doesn't
 890 // allow for message authentication.  Eventually we should fix that.  PLR
 891
 892 void encode_message(Message *msg, uint64_t features, bufferlist& payload)
 893 {
 894   bufferlist front, middle, data;
 895   ceph_msg_footer_old old_footer;
 896   ceph_msg_footer footer;
 897   msg->encode(features, MSG_CRC_ALL);
 898   ::encode(msg->get_header(), payload);
 899
 900   // Here's where we switch to the old footer format.  PLR
 901
 902   footer = msg->get_footer();
 903   old_footer.front_crc = footer.front_crc;
 904   old_footer.middle_crc = footer.middle_crc;
 905   old_footer.data_crc = footer.data_crc;
 906   old_footer.flags = footer.flags;
 907   ::encode(old_footer, payload);
 908
 909   ::encode(msg->get_payload(), payload);
 910   ::encode(msg->get_middle(), payload);
 911   ::encode(msg->get_data(), payload);
 912 }
 913
 914 // See above for somewhat bogus use of the old message footer.  We switch to the current footer
 915 // after decoding the old one so the other form of decode_message() doesn't have to change.
 916 // We've slipped in a 0 signature at this point, so any signature checking after this will
 917 // fail.  PLR
 918
 919 Message *decode_message(CephContext *cct, int crcflags, bufferlist::iterator& p)
 920 {
 921   ceph_msg_header h;
 922   ceph_msg_footer_old fo;
 923   ceph_msg_footer f;
 924   bufferlist fr, mi, da;
 925   ::decode(h, p);
 926   ::decode(fo, p);
 927   f.front_crc = fo.front_crc;
 928   f.middle_crc = fo.middle_crc;
 929   f.data_crc = fo.data_crc;
 930   f.flags = fo.flags;
 931   f.sig = 0;
 932   ::decode(fr, p);
 933   ::decode(mi, p);
 934   ::decode(da, p);
 935   return decode_message(cct, crcflags, h, f, fr, mi, da, nullptr);
 936 }
 937