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