msg: don't truncate message sequence to 32-bits

Message sequence is 64-bits number, but the upper 32-bits get lost
when setting/getting message sequence. This causes messenger to
malfunction after receiving 1^32 messages

Fixes: http://tracker.ceph.com/issues/16122
Signed-off-by: Yan, Zheng <zyan@redhat.com>
(cherry picked from commit d52832d498e2364a38f49bad67d300f0773f889d)

diff --git a/src/msg/Message.h b/src/msg/Message.h
index 9ea453665dc90bd0daa1acf286782ada019b22b0..06b894a06f51c9cce3647518cd39a159874c859c 100644 (file)
--- a/src/msg/Message.h
+++ b/src/msg/Message.h
@@ -423,8 +423,8 @@ public:
   uint64_t get_tid() const { return header.tid; }
   void set_tid(uint64_t t) { header.tid = t; }
 
-  unsigned get_seq() const { return header.seq; }
-  void set_seq(unsigned s) { header.seq = s; }
+  uint64_t get_seq() const { return header.seq; }
+  void set_seq(uint64_t s) { header.seq = s; }
 
   unsigned get_priority() const { return header.priority; }
   void set_priority(__s16 p) { header.priority = p; }

diff --git a/src/msg/async/AsyncConnection.h b/src/msg/async/AsyncConnection.h
index 416bccba42a6318a74fcf6746eb07c7e6bbbbcd1..45ec461fc16ab96ee647307d2d979ac11753be35 100644 (file)
--- a/src/msg/async/AsyncConnection.h
+++ b/src/msg/async/AsyncConnection.h
@@ -278,8 +278,8 @@ class AsyncConnection : public Connection {
   PerfCounters *logger;
   int global_seq;
   __u32 connect_seq, peer_global_seq;
-  atomic_t out_seq;
-  atomic_t ack_left, in_seq;
+  atomic64_t out_seq;
+  atomic64_t ack_left, in_seq;
   int state;
   int state_after_send;
   int sd;

diff --git a/src/msg/simple/Pipe.h b/src/msg/simple/Pipe.h
index d9e346c7e4012e282eb0f37993a4d4cfd2206041..c9a166e6611b8b924505bc4f5a74cd40da193ca1 100644 (file)
--- a/src/msg/simple/Pipe.h
+++ b/src/msg/simple/Pipe.h
@@ -275,7 +275,7 @@ static const int SM_IOV_MAX = (IOV_MAX >= 1024 ? IOV_MAX / 4 : IOV_MAX);
     static const Pipe& Server(int s);
     static const Pipe& Client(const entity_addr_t& pi);
 
-    __u32 get_out_seq() { return out_seq; }
+    uint64_t get_out_seq() { return out_seq; }
 
     bool is_queued() { return !out_q.empty() || send_keepalive || send_keepalive_ack; }
 

diff --git a/src/msg/xio/XioConnection.h b/src/msg/xio/XioConnection.h
index e55ea983197ece090332980458adc0f5d544130b..d62a57e07ed8d5727c311fe70d078d6707a07e12 100644 (file)
--- a/src/msg/xio/XioConnection.h
+++ b/src/msg/xio/XioConnection.h
@@ -93,18 +93,18 @@ private:
     /* XXX */
     uint32_t reconnects;
     uint32_t connect_seq, peer_global_seq;
-    uint32_t in_seq, out_seq_acked; // atomic<uint64_t>, got receipt
-    atomic_t out_seq; // atomic<uint32_t>
+    uint64_t in_seq, out_seq_acked; // atomic<uint64_t>, got receipt
+    atomic64_t out_seq; // atomic<uint32_t>
 
     lifecycle() : state(lifecycle::INIT), reconnects(0), connect_seq(0),
                  peer_global_seq(0), in_seq(0), out_seq_acked(0), 
                  out_seq(0) {}
 
-    void set_in_seq(uint32_t seq) {
+    void set_in_seq(uint64_t seq) {
       in_seq = seq;
     }
 
-    uint32_t next_out_seq() {
+    uint64_t next_out_seq() {
       return out_seq.inc();
     }
 
@@ -139,8 +139,8 @@ private:
 
     uint32_t reconnects;
     uint32_t connect_seq, global_seq, peer_global_seq;
-    uint32_t in_seq, out_seq_acked; // atomic<uint64_t>, got receipt
-    atomic_t out_seq; // atomic<uint32_t>
+    uint64_t in_seq, out_seq_acked; // atomic<uint64_t>, got receipt
+    atomic64_t out_seq; // atomic<uint64_t>
 
     uint32_t flags;
 
@@ -168,11 +168,11 @@ private:
       return startup_state.read();
     }
 
-    void set_in_seq(uint32_t seq) {
+    void set_in_seq(uint64_t seq) {
       in_seq = seq;
     }
 
-    uint32_t next_out_seq() {
+    uint64_t next_out_seq() {
       return out_seq.inc();
     };
 
@@ -329,7 +329,7 @@ typedef boost::intrusive_ptr<XioConnection> XioConnectionRef;
 class XioLoopbackConnection : public Connection
 {
 private:
-  atomic_t seq;
+  atomic64_t seq;
 public:
   explicit XioLoopbackConnection(Messenger *m) : Connection(m->cct, m), seq(0)
     {
@@ -350,11 +350,11 @@ public:
   void mark_down() override {}
   void mark_disposable() override {}
 
-  uint32_t get_seq() {
+  uint64_t get_seq() {
     return seq.read();
   }
 
-  uint32_t next_seq() {
+  uint64_t next_seq() {
     return seq.inc();
   }
 };