Envoy Hot Restart Implementation Analysis

// Increment this whenever there is a shared memory / RPC change that will prevent a hot restart
// from working. Operations code can then cope with this and do a full restart.
const uint64_t HOT_RESTART_VERSION = 11;

/**
 * Shared memory segment. This structure is laid directly into shared memory and is used amongst
 * all running envoy processes.
 */
struct SharedMemory {
  uint64_t size_;
  uint64_t version_;
  pthread_mutex_t log_lock_;
  pthread_mutex_t access_log_lock_;
  std::atomic<uint64_t> flags_;
};
static const uint64_t SHMEM_FLAGS_INITIALIZING = 0x1;

istio-proxy@newsrec-engine-docker-test-ab-3-8865k-mjrm0-6c79c45cb4-8sz26:/$ cat /proc/31/net/unix
Num       RefCount Protocol Flags    Type St Inode Path
000000006f0b8f0a: 00000002 00000000 00000000 0002 01 1895644675 @envoy_domain_socket_parent_0@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
00000000462982b6: 00000002 00000000 00000000 0002 01 1895644674 @envoy_domain_socket_child_0@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@

message HotRestartMessage {
  // Child->parent requests
  message Request {
    message PassListenSocket {
      string address = 1;
    }
    message ShutdownAdmin {
    }
    message Stats {
    }
    message DrainListeners {
    }
    message Terminate {
    }
    oneof request {
      PassListenSocket pass_listen_socket = 1;
      ShutdownAdmin shutdown_admin = 2;
      Stats stats = 3;
      DrainListeners drain_listeners = 4;
      Terminate terminate = 5;
    }
  }

  // Parent->child replies
  message Reply {
    message PassListenSocket {
      int32 fd = 1;
    }
    message ShutdownAdmin {
      uint64 original_start_time_unix_seconds = 1;
    }
    message Span {
      uint32 first = 1;
      uint32 last = 2; // inclusive
    }
    message RepeatedSpan {
      repeated Span spans = 1;
    }
    message Stats {
      // Values for server_stats, which don't fit with the "combination logic" approach.
      uint64 memory_allocated = 1;
      uint64 num_connections = 2;

      // Keys are fully qualified stat names.
      //
      // The amount added to the counter since the last time a message included the counter in this
      // map. (The first time a counter is included in this map, it's the amount added since the
      // final latch() before hot restart began).
      map<string, uint64> counter_deltas = 3;
      // The parent's current values for various gauges in its stats store.
      map<string, uint64> gauges = 4;
      // Maps the string representation of a StatName into an array of Spans,
      // which indicate which of the StatName tokens are dynamic. For example,
      // if we are recording a counter or gauge named "a.b.c.d.e.f", where "a",
      // and "d.e" were created from a StatNameDynamicPool, then we'd map
      // "a.b.c.d.e.f" to the span array [[0,0], [3,4]], where the [0,0] span
      // covers the "a", and the [3,4] span covers "d.e".
      map<string, RepeatedSpan> dynamics = 5;
    }
    oneof reply {
      // When this oneof is of the PassListenSocketReply type, there is a special
      // implied meaning: the recvmsg that got this proto has control data to make
      // the passing of the fd work, so make use of CMSG_SPACE etc.
      PassListenSocket pass_listen_socket = 1;
      ShutdownAdmin shutdown_admin = 2;
      Stats stats = 3;
    }
  }

  oneof requestreply {
    Request request = 1;
    Reply reply = 2;
  }

  bool didnt_recognize_your_last_message = 3;
}

void HotRestartingBase::sendHotRestartMessage(sockaddr_un& address,
                                              const HotRestartMessage& proto) {
  // Child 初始化 Listener, 发现 Listener 的 reuse_port 属性为 false, 则从 parent 获取 listen socket 的 fd 

  // ...

  // send_buf 是 proto message 序列化后的数据, 可能会分成多个 UDP message 来发送

  uint8_t* next_byte_to_send = send_buf.data();
  uint64_t sent = 0;
  while (sent < total_size) {
    const uint64_t cur_chunk_size = std::min(MaxSendmsgSize, total_size - sent);
    iovec iov[1];
    iov[0].iov_base = next_byte_to_send;
    iov[0].iov_len = cur_chunk_size;
    next_byte_to_send += cur_chunk_size;
    sent += cur_chunk_size;
    msghdr message;
    memset(&message, 0, sizeof(message));
    message.msg_name = &address;
    message.msg_namelen = sizeof(address);
    message.msg_iov = iov;
    message.msg_iovlen = 1;

    // 如果是 kPassListenSocket 的消息，则传递 fd

    // Control data stuff, only relevant for the fd passing done with PassListenSocketReply.
    uint8_t control_buffer[CMSG_SPACE(sizeof(int))];
    if (replyIsExpectedType(&proto, HotRestartMessage::Reply::kPassListenSocket) &&
        proto.reply().pass_listen_socket().fd() != -1) {
      memset(control_buffer, 0, CMSG_SPACE(sizeof(int)));
      message.msg_control = control_buffer;
      message.msg_controllen = CMSG_SPACE(sizeof(int));
      cmsghdr* control_message = CMSG_FIRSTHDR(&message);
      control_message->cmsg_level = SOL_SOCKET;
      control_message->cmsg_type = SCM_RIGHTS;
      control_message->cmsg_len = CMSG_LEN(sizeof(int));
      *reinterpret_cast<int*>(CMSG_DATA(control_message)) = proto.reply().pass_listen_socket().fd();
      ASSERT(sent == total_size, "an fd passing message was too long for one sendmsg().");
    }

    sendmsg(my_domain_socket_, &message, 0);
  }

  // ...
}