From f274bbfe689879c81c5a2414a82fd40360e632cc Mon Sep 17 00:00:00 2001
From: Yin Congmin <congmin.yin@intel.com>
Date: Tue, 3 Jan 2023 16:51:04 +0800
Subject: [PATCH] control/discovery: add discovery controller

The discovery contrller implement the basic function.
Use command "python3 -m control.discovery" to start
discovery controller. Client can use command
"nvme discover -t tcp -a ip -s port" to get log pages.

The configuration is in ceph-nvmeof.conf [discovery] part.

feature: https://github.com/ceph/ceph-nvmeof/issues/108

Signed-off-by: Yin Congmin <congmin.yin@intel.com>
---
 ceph-nvmeof.conf     |    5 +
 control/discovery.py | 1110 ++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 1115 insertions(+)
 create mode 100644 control/discovery.py

diff --git a/ceph-nvmeof.conf b/ceph-nvmeof.conf
index c40f065e..e32bcaff 100644
--- a/ceph-nvmeof.conf
+++ b/ceph-nvmeof.conf
@@ -17,6 +17,11 @@ enable_auth = False
 state_update_notify = True
 state_update_interval_sec = 5
 
+[discovery]
+
+addr =
+port = 8009
+
 [ceph]
 
 pool = rbd
diff --git a/control/discovery.py b/control/discovery.py
new file mode 100644
index 00000000..55ecd64d
--- /dev/null
+++ b/control/discovery.py
@@ -0,0 +1,1110 @@
+#
+#  Copyright (c) 2021 International Business Machines
+#  All rights reserved.
+#
+#  SPDX-License-Identifier: LGPL-3.0-or-later
+#
+#  Authors: congmin.yin@intel.com
+#
+
+import argparse
+import grpc
+import json
+import logging
+from .config import GatewayConfig
+from .state import GatewayState, LocalGatewayState, OmapGatewayState, GatewayStateHandler
+
+import rados
+from typing import Dict, Optional
+
+import socket
+import threading
+import time
+import enum
+import struct
+import selectors
+from ctypes import Structure, c_bool, c_ubyte, c_uint8, c_uint16, c_uint32, c_uint64, c_float
+
+# NVMe tcp pdu type
+class NVME_TCP_PDU(enum.IntFlag):
+    ICREQ = 0x0
+    ICRESP = 0x1
+    H2C_TERM = 0x2
+    C2H_TERM = 0x3
+    CMD = 0x4
+    RSP = 0x5
+    H2C_DATA = 0x6
+    C2H_DATA = 0x7
+    TCP_R2T = 0x9
+
+# NVMe tcp opcode
+class NVME_TCP_OPC(enum.IntFlag):
+    DELETE_SQ = 0x0
+    CREATE_SQ = 0x1
+    GET_LOG_PAGE = 0x2
+    DELETE_CQ = 0x4
+    CREATE_CQ = 0x5
+    IDENTIFY = 0x6
+    ABORT = 0x8
+    SET_FEATURES = 0x9
+    GET_FEATURES = 0xa
+    ASYNC_EVE_REQ = 0xc
+    NS_MGMT = 0xd
+    FW_COMMIT = 0x10
+    FW_IMG_DOWNLOAD = 0x11
+    NS_ATTACH = 0x15
+    KEEP_ALIVE = 0x18
+    FABRIC_TYPE = 0x7F
+
+# NVMe tcp fabric command type
+class NVME_TCP_FCTYPE(enum.IntFlag):
+    PROP_SET = 0x0
+    CONNECT = 0x1
+    PROP_GET = 0x4
+    AUTH_SEND = 0x5
+    AUTH_RECV = 0x6
+    DISCONNECT = 0x8
+
+# NVMe controller register space offsets
+class NVME_CTL(enum.IntFlag):
+    CAPABILITIES = 0x0
+    VERSION = 0x08
+    CONFIGURATION = 0x14
+    STATUS = 0x1c
+
+
+# NVM subsystem types
+class NVMF_SUBTYPE(enum.IntFlag):
+    # Discovery type for NVM subsystem
+    DISCOVERY = 0x1
+    # NVMe type for NVM subsystem
+    NVME = 0x2
+
+# NVMe over Fabrics transport types
+transport_types = {
+    "RDMA": 0x1,
+    "FC": 0x2,
+    "TCP": 0x3,
+    "INTRA_HOST": 0xfe
+}
+
+# Address family types
+adrfam_types = {
+    "ipv4": 0x1,
+    "ipv6": 0x2,
+    "ib": 0x3,
+    "fc": 0x4,
+    "intra_host": 0xfe
+}
+
+# Transport requirement, secure channel requirements
+# Connections shall be made over a fabric secure channel
+class NVMF_TREQ_SECURE_CHANNEL(enum.IntFlag):
+    NOT_SPECIFIED = 0x0
+    REQUIRED = 0x1
+    NOT_REQUIRED = 0x2
+
+# maximum number of connections
+MAX_CONNECTION = 10240
+
+# NVMe tcp package length, refer: MTU = 1500 bytes
+NVME_TCP_PDU_UNIT = 1024
+
+# Max SQ head pointer
+SQ_HEAD_MAX = 128
+
+lock = threading.Lock()
+
+# Global controller id
+GLOBAL_CNLID = 0x1
+
+# Global generation counter
+GLOBAL_GEN_CNT = 0x1
+
+class Connection(Structure):
+    """Data used multiple times in each connection."""
+
+    _fields_ = [
+        ("nvmeof_connect_data_hostid", c_ubyte * 16),
+        ("nvmeof_connect_data_cntlid", c_uint16),
+        ("nvmeof_connect_data_subnqn", c_ubyte * 256),
+        ("nvmeof_connect_data_hostnqn", c_ubyte * 256),
+        ("sq_head_ptr", c_uint16),
+        ("unsent_log_page_len", c_uint32),
+        ("property_data", c_ubyte * 8),
+        ("property_set", c_bool),
+        ("shutdown_now", c_bool),
+        ("cnlid", c_uint16),
+        ("gen_cnt", c_uint16),
+        ("recv_async", c_bool),
+        ("async_cmd_id", c_uint16),
+        ("keep_alive_time", c_float),
+        ("keep_alive_timeout", c_uint32),
+    ]
+
+    def __init__(self):
+        self.connection = 0
+        self.allow_listeners = []
+        self.log_page = bytearray()
+        self.recv_buffer = b''
+
+        self.unsent_log_page_len = 0
+        self.property_set = False
+        self.shutdown_now = False
+        self.recv_async = False
+
+class Pdu(Structure):
+    _fields_ = [
+        ("type", c_uint8),
+        ("specical_flag", c_uint8),
+        ("header_length", c_uint8),
+        ("data_offset", c_uint8),
+        ("packet_length", c_uint32),
+    ]
+
+    def compose_reply(self):
+        return bytes(self)
+
+class ICResp(Structure):
+    _fields_ = [
+        # pdu version format
+        ("version_format", c_uint16),
+        # controller Pdu data alignment
+        ("data_alignment", c_uint8),
+        # digest types enabled
+        ("digest_types", c_uint8),
+        # Maximum data capsules per r2t supported
+        ("maximum_data_capsules", c_uint32),
+    ]
+
+    def compose_reply(self):
+        return bytes(self)
+
+class CqeConnect(Structure):
+    _fields_ = [
+        ("controller_id", c_uint16),
+        ("authentication", c_uint16),
+        ("reserved", c_uint32),
+        ("sq_head_ptr", c_uint16),
+        ("sq_id", c_uint16),
+        ("cmd_id", c_uint16),
+        ("status", c_uint16)
+    ]
+
+    def compose_reply(self):
+        return bytes(self)
+
+class CqePropertyGetSet(Structure):
+    _fields_ = [
+        # property data for property get, reserved for property set
+        ("property_data", c_ubyte * 8),
+        ("sq_head_ptr", c_uint16),
+        ("sq_id", c_uint16),
+        ("cmd_id", c_uint16),
+        ("status", c_uint16)
+    ]
+
+    def compose_reply(self):
+        return bytes(self)
+
+class NVMeTcpDataPdu(Structure):
+    _fields_ = [
+        ("cmd_id", c_uint16),
+        ("transfer_tag", c_uint16),
+        ("data_offset", c_uint32),
+        ("data_length", c_uint32),
+        ("reserved", c_uint32)
+    ]
+
+    def compose_reply(self):
+        return bytes(self)
+
+class NVMeIdentify(Structure):
+    _fields_ = [
+        # skip some fields, include VID, SSVID, SN, MN
+        ("todo_fields1", c_ubyte * 64),
+        ("firmware_revision", c_ubyte * 8),
+        # RAB, IEEE, CMIC
+        ("todo_fields2", c_ubyte * 5),
+        # maximum data transfer size
+        ("mdts", c_uint8),
+        ("controller_id", c_uint16),
+        ("version", c_uint8 * 4),
+        # RTD3R, RTD3E
+        ("todo_fields3", c_ubyte * 8),
+        # optional asynchronous events supported
+        ("oaes", c_ubyte * 4),
+        # CTRATT, RRLS, CNTRLTYPE, FGUID, NVMe Management Interface, OACS, ACL
+        ("todo_fields4", c_ubyte * 163),
+        # asynchronous events request limit
+        ("aerl", c_uint8),
+        ("firmware_updates", c_uint8),
+        # log page attributes
+        ("lpa", c_uint8),
+        # error log page entries(ELPE)
+        ("elpe", c_uint8),
+        # NPSS, AVSCC, APSTA, WCTEMP, CCTEMP, MTFA, HMPRE, HMIN, TNVMCAP...
+        # TODO: keep alive support - timer value(KAS)?
+        ("todo_fields5", c_ubyte * 251),
+        # maximum outstanding commands
+        ("max_cmd", c_uint16),
+        # number of namespace, optional NVM command support
+        ("todo_fields6", c_uint8 * 6),
+        # fused operation support
+        ("fused_operation", c_uint16),
+        # FNA, VWC, AWUN, AWUPF, NVSCC, NWPC
+        ("todo_fields7", c_uint8 * 8),
+        # atomic compare & write unit
+        ("acwu", c_uint16),
+        ("reserved1", c_uint16),
+        # SGL support
+        ("sgls", c_uint8 * 4),
+        # maxinum number of allowed namespaces
+        ("mnan", c_uint32),
+        ("reserved2", c_ubyte * 224),
+        ("subnqn", c_ubyte * 256),
+        ("reserved3", c_ubyte * 768),
+        ("nvmeof_attributes", c_ubyte * 256),
+        ("power_state_attributes", c_ubyte * 1024),
+        ("vendor_specific", c_ubyte * 1024)
+    ]
+
+    def compose_reply(self):
+        return bytes(self)
+
+# for set feature, keep alive and async
+class  CqeNVMe(Structure):
+    _fields_ = [
+        ("dword0", c_uint32),
+        ("dword1", c_uint32),
+        ("sq_head_ptr", c_uint16),
+        ("sq_id", c_uint16),
+        ("cmd_id", c_uint16),
+        ("status", c_uint16)
+    ]
+
+    def compose_reply(self):
+        return bytes(self)
+
+class NVMeGetLogPage(Structure):
+    _fields_ = [
+        # generation counter
+        ("genctr", c_uint64),
+        # number of records
+        ("numrec", c_uint64),
+        #record format
+        ("recfmt", c_uint16),
+        ("reserved", c_ubyte * 1006)
+    ]
+
+    def compose_short_reply(self):
+        return bytes(self)[:16]
+
+    def compose_data_reply(self):
+        return bytes(self)
+
+class DiscoveryLogEntry(Structure):
+    _fields_ = [
+        ("trtype", c_uint8),
+        ("adrfam", c_uint8),
+        ("subtype", c_uint8),
+        ("treq", c_uint8),
+        ("port_id", c_uint16),
+        ("controller_id", c_uint16),
+        # admin max SQ size
+        ("asqsz", c_uint16),
+        ("reserved1", c_ubyte * 22),
+        ("trsvcid", c_ubyte * 32),
+        ("reserved2", c_ubyte * 192),
+        ("subnqn", c_ubyte * 256),
+        ("traddr", c_ubyte * 256),
+        # Transport specific address subtype
+        ("tsas", c_ubyte * 256)
+    ]
+
+    def compose_reply(self):
+        return bytes(self)
+
+class DiscoveryService:
+    """Implements discovery controller.
+
+    Response discover request from initiator.
+
+    Instance attributes:
+        version: Discovery controller version
+        config: Basic gateway parameters
+        logger: Logger instance to track discovery controller events
+        omap_name: OMAP object name
+        ioctx: I/O context which allows OMAP access
+        discovery_addr: Discovery controller addr which allows initiator send command
+        discovery_port: Discovery controller's listening port
+    """
+
+    BDEV_PREFIX = "bdev_"
+    NAMESPACE_PREFIX = "namespace_"
+    SUBSYSTEM_PREFIX = "subsystem_"
+    HOST_PREFIX = "host_"
+    LISTENER_PREFIX = "listener_"
+
+    def __init__(self, config):
+        self.logger = logging.getLogger(__name__)
+        self.logger.setLevel(level=logging.DEBUG)
+
+        self.version = 1
+        self.config = config
+
+        gateway_group = self.config.get("gateway", "group")
+        self.omap_name = f"nvmeof.{gateway_group}.state" \
+            if gateway_group else "nvmeof.state"
+        self.logger.info(f"log pages info from omap: {self.omap_name}")
+
+        ceph_pool = self.config.get("ceph", "pool")
+        ceph_conf = self.config.get("ceph", "config_file")
+        conn = rados.Rados(conffile=ceph_conf)
+        conn.connect()
+        self.ioctx = conn.open_ioctx(ceph_pool)
+
+        self.discovery_addr = self.config.get("discovery", "addr")
+        self.discovery_port = self.config.get("discovery", "port")
+        if self.discovery_addr == '' or self.discovery_port == '':
+            self.logger.error("discovery addr/port are empty.")
+            assert 0
+        self.logger.info(f"discovery addr: {self.discovery_addr} port: {self.discovery_port}")
+
+        self.conn_vals = {}
+        self.selector = selectors.DefaultSelector()
+
+    def _read_all(self) -> Dict[str, str]:
+        """Reads OMAP and returns dict of all keys and values."""
+
+        with rados.ReadOpCtx() as read_op:
+            iter, _ = self.ioctx.get_omap_vals(read_op, "", "", -1)
+            self.ioctx.operate_read_op(read_op, self.omap_name)
+            omap_dict = dict(iter)
+        return omap_dict
+
+    def _get_vals(self, omap_dict, prefix):
+        """Read values from the OMAP dict."""
+
+        vals = []
+        for (key, val) in omap_dict.items():
+            if key.startswith(prefix):
+                val_text = val.decode('utf-8')
+                js = json.loads(val_text)
+                vals.append(js)
+        return vals
+
+    def reply_initialize(self, conn):
+        """Reply initialize request."""
+
+        self.logger.debug("handle ICreq.")
+        pdu_reply = Pdu()
+        pdu_reply.type = NVME_TCP_PDU.ICRESP
+        pdu_reply.header_length = 128
+        pdu_reply.packet_length = 128
+
+        icresp_reply = ICResp()
+        icresp_reply.maximum_data_capsules = 131072
+
+        reply = bytes(pdu_reply) + bytes(icresp_reply) + bytes(112)
+
+        try:
+            conn.sendall(reply)
+        except BrokenPipeError:
+            self.logger.error("client disconnected unexpectedly.")
+            return -1
+        self.logger.debug("reply initialize connection request.")
+        return 0
+
+    def reply_fc_cmd_connect(self, conn, data, cmd_id):
+        """Reply connect request."""
+
+        self.logger.debug("handle connect request.")
+        self_conn = self.conn_vals[conn.fileno()]
+        hf_nvmeof_cmd_connect_rsvd1 = struct.unpack_from('<19B', data, 13)
+        SIGL1 = struct.unpack_from('<QI4B', data, 32)
+        address = SIGL1[0]
+        length = SIGL1[1]
+        reserved3 = SIGL1[2]
+        descriptor_type = SIGL1[5]
+
+        CMD2 = struct.unpack_from('<HHHBBI', data, 48)
+        record_format = CMD2[0]
+        queue_id = CMD2[1]
+        submission_queue_size = CMD2[2]
+        connect_attributes = CMD2[3]
+        keep_alive_timeout = CMD2[5]
+
+        self_conn =self.conn_vals[conn.fileno()]
+        self_conn.keep_alive_time = time.time()
+        self.logger.debug(f"connection keep alive {keep_alive_timeout}ms.")
+        self_conn.keep_alive_timeout = keep_alive_timeout
+
+        self_conn.nvmeof_connect_data_hostid = \
+            struct.unpack_from('<16B', data, 72)
+        self_conn.nvmeof_connect_data_cntlid = \
+            struct.unpack_from('<H', data, 88)[0]
+        self_conn.nvmeof_connect_data_subnqn = \
+            struct.unpack_from('<256B', data, 328)
+        self_conn.nvmeof_connect_data_hostnqn = \
+            struct.unpack_from('<256B', data, 584)
+
+        pdu_reply = Pdu()
+        pdu_reply.type = NVME_TCP_PDU.RSP
+        pdu_reply.header_length = 24
+        pdu_reply.packet_length = 24
+
+        connect_reply = CqeConnect()
+        connect_reply.controller_id = self_conn.cnlid
+        connect_reply.sq_head_ptr = self_conn.sq_head_ptr
+
+        reply = bytes(pdu_reply) + bytes(connect_reply)
+        try:
+            conn.sendall(reply)
+        except BrokenPipeError:
+            self.logger.error("client disconnected unexpectedly.")
+            return -1
+        self.logger.debug("reply connect request.")
+        return 0
+
+    def reply_fc_cmd_prop_get(self, conn, data, cmd_id):
+        """Reply property get request."""
+
+        self.logger.debug("handle property get request.")
+        self_conn = self.conn_vals[conn.fileno()]
+        nvmeof_prop_get_set_rsvd0 = struct.unpack_from('<35B', data, 13)
+        # property size = (attrib+1)x4, 0x1 means 8 bytes
+        nvmeof_prop_get_set_attrib = struct.unpack_from('<1B', data, 48)[0]
+        nvmeof_prop_get_set_rsvd1 = struct.unpack_from('<3B', data, 49)
+        nvmeof_prop_get_set_offset = struct.unpack_from('<I', data, 52)[0]
+
+        pdu_reply = Pdu()
+        pdu_reply.type = NVME_TCP_PDU.RSP
+        pdu_reply.header_length = 24
+        pdu_reply.packet_length = 24
+
+        # reply different property data
+        property_get = CqePropertyGetSet()
+        if NVME_CTL(nvmeof_prop_get_set_offset) == NVME_CTL.CAPABILITIES:
+            # controller capabilities
+            # \x7f = maxinum queue entries support:128
+            # \x01 contiguous queues required: true
+            # \x1e timeout(to ready status): 1e(15000ms default in server side)
+            # \x20 command sets supportd: 1 (NVM IO command set)?
+            property_get.property_data = (c_ubyte * 8)(0x7f, 0x00, \
+                0x01, 0x1e, 0x20, 0x00, 0x00, 0x00)
+        elif NVME_CTL(nvmeof_prop_get_set_offset) == NVME_CTL.CONFIGURATION:
+            # b'\x00\x00\x46\x00\x00\x00\x00\x00'
+            # 0x46: IO Submission Queue Entry Size: 0x6 (64 bytes)
+            # IO Completion Queue Entry Size: 0x4 (16 bytes)
+            property_get.property_data = self_conn.property_data
+        elif NVME_CTL(nvmeof_prop_get_set_offset) == NVME_CTL.STATUS:
+            shutdown_notification = (self_conn.property_data[1] >> 6) & 0x3
+            if shutdown_notification == 0:
+                if self_conn.property_set is True:
+                    # controller status: ready
+                    property_get.property_data = (c_ubyte * 8)(0x01, 0x00, \
+                        0x00, 0x00, 0x00, 0x00, 0x00, 0x00)
+                else:
+                    property_get.property_data = (c_ubyte * 8)(0x00, 0x00, \
+                        0x00, 0x00, 0x00, 0x00, 0x00, 0x00)
+            else:
+                # here shutdown_notification should be 0x1
+                property_get.property_data = (c_ubyte * 8)(0x09, 0x00, \
+                    0x00, 0x00, 0x00, 0x00, 0x00, 0x00)
+                self_conn.shutdown_now = True
+        elif NVME_CTL(nvmeof_prop_get_set_offset) == NVME_CTL.VERSION:
+            # nvme version: 1.3
+            property_get.property_data = (c_ubyte * 8)(0x00, 0x03, \
+                0x01, 0x00, 0x00, 0x00, 0x00, 0x00)
+        else:
+            self.logger.error("unsupported type for property getting.")
+        property_get.sq_head_ptr = self_conn.sq_head_ptr
+        property_get.cmd_id = cmd_id
+
+        reply = bytes(pdu_reply) + bytes(property_get)
+        try:
+            conn.sendall(reply)
+        except BrokenPipeError:
+            self.logger.error("client disconnected unexpectedly.")
+            return -1
+        self.logger.debug("reply property get request.")
+        return 0
+
+    def reply_fc_cmd_prop_set(self, conn, data, cmd_id):
+        """Reply property set request."""
+
+        self.logger.debug("handle property set request.")
+        self_conn = self.conn_vals[conn.fileno()]
+        nvmeof_prop_get_set_rsvd0 = struct.unpack_from('<35B', data, 13)
+        nvmeof_prop_get_set_attrib = struct.unpack_from('<1B', data, 48)[0]
+        nvmeof_prop_get_set_rsvd1 = struct.unpack_from('<3B', data, 49)
+        nvmeof_prop_get_set_offset = struct.unpack_from('<I', data, 52)[0]
+
+        if NVME_CTL(nvmeof_prop_get_set_offset) == NVME_CTL.CAPABILITIES:
+            self.logger.error("property setting of capabilities is not supported.")
+        elif NVME_CTL(nvmeof_prop_get_set_offset) == NVME_CTL.CONFIGURATION:
+            self_conn.property_data = struct.unpack_from('<8B', data, 56)
+            self_conn.property_set = True
+        elif NVME_CTL(nvmeof_prop_get_set_offset) == NVME_CTL.STATUS:
+            self.logger.error("property setting of status is not supported.")
+        elif NVME_CTL(nvmeof_prop_get_set_offset) == NVME_CTL.VERSION:
+            self.logger.error("property setting of version is not supported.")
+        else:
+            self.logger.error("unsupported type for property settings.")
+
+        pdu_reply = Pdu()
+        pdu_reply.type = NVME_TCP_PDU.RSP
+        pdu_reply.header_length = 24
+        pdu_reply.packet_length = 24
+
+        property_set = CqePropertyGetSet()
+        property_set.sq_head_ptr = self_conn.sq_head_ptr
+        property_set.cmd_id = cmd_id
+
+        reply = bytes(pdu_reply) + bytes(property_set)
+        try:
+            conn.sendall(reply)
+        except BrokenPipeError:
+            self.logger.error("client disconnected unexpectedly.")
+            return -1
+        self.logger.debug("reply property set request.")
+        return 0
+
+    def reply_identify(self, conn, data, cmd_id):
+        """Reply identify request."""
+
+        self.logger.debug("handle identify request.")
+        self_conn = self.conn_vals[conn.fileno()]
+        nvme_nsid = struct.unpack_from('<I', data, 12)[0]
+        nvme_rsvd1 = struct.unpack_from('<Q', data, 16)[0]
+        nvme_mptr = struct.unpack_from('<Q', data, 24)[0]
+        nvme_sgl = struct.unpack_from('<16B', data, 32)
+        nvme_sgl_desc_type = nvme_sgl[15] & 0xF0
+        nvme_sgl_desc_sub_type = nvme_sgl[15] & 0x0F
+        nvme_identify_dword10 = struct.unpack_from('<I', data, 48)[0]
+        nvme_identify_dword11 = struct.unpack_from('<I', data, 52)[0]
+        nvme_identify_dword12 = struct.unpack_from('<I', data, 56)[0]
+        nvme_identify_dword13 = struct.unpack_from('<I', data, 60)[0]
+        nvme_identify_dword14 = struct.unpack_from('<I', data, 64)[0]
+        nvme_identify_dword15 = struct.unpack_from('<I', data, 68)[0]
+
+        pdu_reply = Pdu()
+        pdu_reply.type = NVME_TCP_PDU.C2H_DATA
+        # 0x0c == 0b1100, means pdu data last: set, pdu data success: set
+        pdu_reply.specical_flag = 0x0c
+        pdu_reply.header_length = 24
+        pdu_reply.data_offset = 24
+        pdu_reply.packet_length = 4120
+
+        nvme_tcp_data_pdu = NVMeTcpDataPdu()
+        nvme_tcp_data_pdu.cmd_id = cmd_id
+        nvme_tcp_data_pdu.data_length = 4096
+
+        identify_reply = NVMeIdentify()
+        # version: 0.01
+        identify_reply.firmware_revision = (c_ubyte * 8)(0x30, 0x30, \
+            0x2e, 0x30, 0x31, 0x20, 0x20, 0x20)
+        # maximum data transfer size: 2^5=32 pages
+        identify_reply.mdts = 0x05
+        identify_reply.controller_id = self_conn.cnlid
+        # version: 1.3, TODO: get from hardware
+        identify_reply.version = (c_uint8 * 4)(0x00, 0x30, 0x01, 0x00)
+        identify_reply.oaes = (c_ubyte * 4)(0x00, 0x00, 0x00, 0x80)
+        # asynchronous events request limit: 4 events(3+1)
+        identify_reply.aerl = 0x03
+        # log page attributes: Extended data get log page support: True
+        identify_reply.lpa = 0x04
+        # error log page entries:128 entries(127+1)
+        identify_reply.elpe = 0x7f
+        identify_reply.max_cmd = 128
+        identify_reply.fused_operation = 0x0001
+        identify_reply.sgls = (c_uint8 * 4)(0x05, 0x00, 0x10, 0x00)
+        for i in range(256):
+            identify_reply.subnqn[i] = self_conn.nvmeof_connect_data_subnqn[i]
+
+        reply = bytes(pdu_reply) + bytes(nvme_tcp_data_pdu) + \
+                bytes(identify_reply)
+        try:
+            conn.sendall(reply)
+        except BrokenPipeError:
+            self.logger.error("client disconnected unexpectedly.")
+            return -1
+        self.logger.debug("reply identify request.")
+        return 0
+
+    def reply_set_feature(self, conn, data, cmd_id):
+        """Reply set feature request."""
+
+        self.logger.debug("handle set feature request.")
+        self_conn = self.conn_vals[conn.fileno()]
+        nvme_nsid = struct.unpack_from('<I', data, 12)[0]
+        nvme_rsvd1 = struct.unpack_from('<Q', data, 16)[0]
+        nvme_mptr = struct.unpack_from('<Q', data, 24)[0]
+        nvme_sgl = struct.unpack_from('<16B', data, 32)
+        nvme_sgl_desc_type = nvme_sgl[15] & 0xF0
+        nvme_sgl_desc_sub_type = nvme_sgl[15] & 0x0F
+        # dword10 may include Feature Identifier:
+        # Asynchronous Event Configuration (0x0b) (Not currently used)
+        nvme_set_features_dword10 = struct.unpack_from('<I', data, 48)[0]
+        nvme_set_features_dword11 = struct.unpack_from('<I', data, 52)[0]
+        nvme_set_features_dword12 = struct.unpack_from('<I', data, 56)[0]
+        nvme_set_features_dword13 = struct.unpack_from('<I', data, 60)[0]
+        nvme_set_features_dword14 = struct.unpack_from('<I', data, 64)[0]
+        nvme_set_features_dword15 = struct.unpack_from('<I', data, 68)[0]
+
+        pdu_reply = Pdu()
+        pdu_reply.type = NVME_TCP_PDU.RSP
+        pdu_reply.header_length = 24
+        pdu_reply.packet_length = 24
+
+        set_feature_reply =  CqeNVMe()
+        set_feature_reply.sq_head_ptr = self_conn.sq_head_ptr
+        set_feature_reply.cmd_id = cmd_id
+
+        reply = bytes(pdu_reply) + bytes(set_feature_reply)
+        try:
+            conn.sendall(reply)
+        except BrokenPipeError:
+            self.logger.error("client disconnected unexpectedly.")
+            return -1
+        self.logger.debug("reply set feature request.")
+        return 0
+
+    def reply_get_feature(self, conn, data, cmd_id):
+        """Reply get feature request."""
+
+        self.logger.debug("handle get feature request.")
+        self_conn = self.conn_vals[conn.fileno()]
+        nvme_nsid = struct.unpack_from('<I', data, 12)[0]
+        nvme_rsvd1 = struct.unpack_from('<Q', data, 16)[0]
+        nvme_mptr = struct.unpack_from('<Q', data, 24)[0]
+        nvme_sgl = struct.unpack_from('<16B', data, 32)
+        nvme_sgl_desc_type = nvme_sgl[15] & 0xF0
+        nvme_sgl_desc_sub_type = nvme_sgl[15] & 0x0F
+        nvme_get_features_dword10 = struct.unpack_from('<I', data, 48)[0]
+        nvme_get_features_dword11 = struct.unpack_from('<I', data, 52)[0]
+        nvme_get_features_dword12 = struct.unpack_from('<I', data, 56)[0]
+        nvme_get_features_dword13 = struct.unpack_from('<I', data, 60)[0]
+        nvme_get_features_dword14 = struct.unpack_from('<I', data, 64)[0]
+        nvme_get_features_dword15 = struct.unpack_from('<I', data, 68)[0]
+
+        pdu_reply = Pdu()
+        pdu_reply.type = NVME_TCP_PDU.RSP
+        pdu_reply.header_length = 24
+        pdu_reply.packet_length = 24
+
+        get_feature_reply =  CqeNVMe()
+        get_feature_reply.dword0 = self_conn.keep_alive_timeout
+        get_feature_reply.sq_head_ptr = self_conn.sq_head_ptr
+        get_feature_reply.cmd_id = cmd_id
+
+        reply = bytes(pdu_reply) + bytes(get_feature_reply)
+        try:
+            conn.sendall(reply)
+        except BrokenPipeError:
+            self.logger.error("client disconnected unexpectedly.")
+            return -1
+        self.logger.debug("reply get feature request.")
+        return 0
+
+    def reply_get_log_page(self, conn, data, cmd_id):
+        """Reply get log page request."""
+
+        self.logger.debug("handle get log page request.")
+        self_conn = self.conn_vals[conn.fileno()]
+        my_omap_dict = self._read_all()
+        listeners = self._get_vals(my_omap_dict, self.LISTENER_PREFIX)
+        hosts = self._get_vals(my_omap_dict, self.HOST_PREFIX)
+        hostnqn = "".join(chr(byte) for byte \
+            in self_conn.nvmeof_connect_data_hostnqn[:256]).rstrip('\x00')
+
+        nvme_nsid = struct.unpack_from('<I', data, 12)[0]
+        nvme_rsvd1 = struct.unpack_from('<Q', data, 16)[0]
+        nvme_mptr = struct.unpack_from('<Q', data, 24)[0]
+        nvme_sgl = struct.unpack_from('<16B', data, 32)
+        nvme_sgl_desc_type = nvme_sgl[15] & 0xF0
+        nvme_sgl_desc_sub_type = nvme_sgl[15] & 0x0F
+        nvme_get_logpage_dword10 = struct.unpack_from('<I', data, 48)[0]
+        # nvme_get_logpage_numd indicate the reply bytes, rule: (values+1)*4
+        nvme_get_logpage_numd = struct.unpack_from('<I', data, 50)[0]
+        nvme_data_len = (nvme_get_logpage_numd + 1) * 4
+        nvme_get_logpage_dword11 = struct.unpack_from('<I', data, 52)[0]
+        # Logpage offset overlaps with dword13
+        nvme_logpage_offset = struct.unpack_from('<Q', data, 56)[0]
+        nvme_get_logpage_dword13 = struct.unpack_from('<I', data, 60)[0]
+        nvme_get_logpage_dword14 = struct.unpack_from('<I', data, 64)[0]
+        nvme_get_logpage_dword15 = struct.unpack_from('<I', data, 68)[0]
+        get_logpage_lid = nvme_get_logpage_dword10 & 0xFF
+        get_logpage_lsp = (nvme_get_logpage_dword10 >> 8) & 0x1F
+        get_logpage_lsi = nvme_get_logpage_dword11 >> 16
+        get_logpage_uid_idx = nvme_get_logpage_dword14 & 0x3F
+
+        if get_logpage_lid != 0x70:
+            self.logger.error("request type error, not discovery request.")
+            return -1
+
+        # Filter listeners based on host access permissions
+        allow_listeners = self_conn.allow_listeners
+        if len(allow_listeners) == 0:
+            for host in hosts:
+                a = host["host_nqn"]
+                if host["host_nqn"] == '*' or host["host_nqn"] == hostnqn:
+                    for listener in listeners:
+                        # TODO: It is better to change nqn in the "listener"
+                        # to subsystem_nqn to avoid confusion
+                        if host["subsystem_nqn"] == listener["nqn"]:
+                            allow_listeners += [listener,]
+            self_conn.allow_listeners = allow_listeners
+
+        # Prepare all log page data segments
+        if self_conn.unsent_log_page_len == 0 and nvme_data_len > 16:
+            self_conn.unsent_log_page_len = 1024 * (len(allow_listeners) + 1)
+            self_conn.log_page = bytearray(self_conn.unsent_log_page_len)
+
+            nvme_get_log_page_reply = NVMeGetLogPage()
+            nvme_get_log_page_reply.genctr = self_conn.gen_cnt
+            nvme_get_log_page_reply.numrec = len(allow_listeners)
+            self_conn.log_page[0:1024] = bytes(nvme_get_log_page_reply)
+
+            # log entries
+            log_entry_counter = 0
+            while log_entry_counter < len(allow_listeners):
+                log_entry = DiscoveryLogEntry()
+                trtype = transport_types.get(allow_listeners[log_entry_counter]["trtype"])
+                if trtype is None:
+                    self.logger.error("unsupported transport type")
+                else:
+                    log_entry.trtype = trtype
+                adrfam = adrfam_types.get(allow_listeners[log_entry_counter]["adrfam"])
+                if adrfam is None:
+                    self.logger.error("unsupported adress family")
+                else:
+                    log_entry.adrfam = adrfam
+                log_entry.subtype = NVMF_SUBTYPE.NVME
+                log_entry.treq = NVMF_TREQ_SECURE_CHANNEL.NOT_REQUIRED
+                log_entry.port_id = log_entry_counter
+                log_entry.controller_id = 0xffff
+                log_entry.asqsz = 128
+                # transport service indentifier
+                log_entry.trsvcid = (c_ubyte * 32)(*[c_ubyte(x) for x \
+                    in allow_listeners[log_entry_counter]["trsvcid"].encode()])
+                log_entry.trsvcid[len(allow_listeners[log_entry_counter]["trsvcid"]):] = \
+                    [c_ubyte(0x20)] * (32 - len(allow_listeners[log_entry_counter]["trsvcid"]))
+                # NVM subsystem qualified name
+                log_entry.subnqn = (c_ubyte * 256)(*[c_ubyte(x) for x \
+                    in allow_listeners[log_entry_counter]["nqn"].encode()])
+                log_entry.subnqn[len(allow_listeners[log_entry_counter]["nqn"]):] = \
+                    [c_ubyte(0x00)] * (256 - len(allow_listeners[log_entry_counter]["nqn"]))
+                # Transport address
+                log_entry.traddr = (c_ubyte * 256)(*[c_ubyte(x) for x \
+                    in allow_listeners[log_entry_counter]["traddr"].encode()])
+                log_entry.traddr[len(allow_listeners[log_entry_counter]["traddr"]):] = \
+                    [c_ubyte(0x20)] * (256 - len(allow_listeners[log_entry_counter]["traddr"]))
+
+                self_conn.log_page[1024*(log_entry_counter+1): \
+                    1024*(log_entry_counter+2)] = bytes(log_entry)
+                log_entry_counter += 1
+        else:
+            self.logger.debug("in the process of sending log pages...")
+
+        reply = b''
+        pdu_and_nvme_pdu_len = 8 + 16
+        pdu_reply = Pdu()
+        pdu_reply.type = NVME_TCP_PDU.C2H_DATA
+        pdu_reply.specical_flag = 0x0c
+        pdu_reply.header_length = pdu_and_nvme_pdu_len
+        pdu_reply.data_offset = pdu_and_nvme_pdu_len
+        pdu_reply.packet_length = pdu_and_nvme_pdu_len + nvme_data_len
+
+        nvme_tcp_data_pdu = NVMeTcpDataPdu()
+        nvme_tcp_data_pdu.cmd_id = cmd_id
+        nvme_tcp_data_pdu.data_length = nvme_data_len
+
+        # reply based on the received get log page request packet(length)
+        if nvme_data_len == 16:
+            # nvme cli version: 1.x
+            nvme_get_log_page_reply = NVMeGetLogPage()
+            nvme_get_log_page_reply.genctr = self_conn.gen_cnt
+            nvme_get_log_page_reply.numrec = len(listeners)
+
+            reply = bytes(pdu_reply) + bytes(nvme_tcp_data_pdu) + \
+                    bytes(nvme_get_log_page_reply)[:16]
+        elif nvme_data_len == 1024 and nvme_logpage_offset == 0:
+            # nvme cli version: 2.x
+            nvme_get_log_page_reply = NVMeGetLogPage()
+            nvme_get_log_page_reply.genctr = self_conn.gen_cnt
+            nvme_get_log_page_reply.numrec = len(listeners)
+
+            reply = bytes(pdu_reply) + bytes(nvme_tcp_data_pdu) + \
+                    bytes(nvme_get_log_page_reply)
+        elif nvme_data_len % 1024 == 0:
+            # reply log pages
+            reply = bytes(pdu_reply) + bytes(nvme_tcp_data_pdu) + \
+                self_conn.log_page[nvme_logpage_offset:nvme_logpage_offset+nvme_data_len]
+            self_conn.unsent_log_page_len -= nvme_data_len
+            if self_conn.unsent_log_page_len == 0:
+                self_conn.log_page = b''
+                self_conn.property_set = False
+                self_conn.allow_listeners = []
+        else:
+            self.logger.error("lenghth error. It need to be 16 or n*1024")
+            return -1
+        try:
+            conn.sendall(reply)
+        except BrokenPipeError:
+            self.logger.error("client disconnected unexpectedly.")
+            return -1
+        self.logger.debug("reply get log page request.")
+        return 0
+
+    def reply_keep_alive(self, conn, data, cmd_id):
+        """Reply keep alive request."""
+
+        self.logger.debug("handle keep alive request.")
+        self_conn = self.conn_vals[conn.fileno()]
+        self_conn.keep_alive_time = time.time()
+        nvme_sgl = struct.unpack_from('<16B', data, 32)
+        nvme_sgl_desc_type = nvme_sgl[15] & 0xF0
+        nvme_sgl_desc_sub_type = nvme_sgl[15] & 0x0F
+        nvme_keep_alive_dword10 = struct.unpack_from('<I', data, 48)[0]
+        nvme_keep_alive_dword11 = struct.unpack_from('<I', data, 52)[0]
+        nvme_keep_alive_dword12 = struct.unpack_from('<I', data, 56)[0]
+        nvme_keep_alive_dword13 = struct.unpack_from('<I', data, 60)[0]
+        nvme_keep_alive_dword14 = struct.unpack_from('<I', data, 64)[0]
+        nvme_keep_alive_dword15 = struct.unpack_from('<I', data, 68)[0]
+
+        pdu_reply = Pdu()
+        pdu_reply.type = NVME_TCP_PDU.RSP
+        pdu_reply.header_length = b'\x18'
+        pdu_reply.packet_length = b'\x18\x00\x00\x00'
+
+         # Cqe for keep alive
+        keep_alive_reply = CqeNVMe()
+        keep_alive_reply.sq_head_ptr = struct.pack('<H', self_conn.sq_head_ptr)
+        keep_alive_reply.cmd_id = struct.pack('<H', cmd_id)
+
+        reply = pdu_reply.compose_reply() + keep_alive_reply.compose_reply()
+        try:
+            conn.sendall(reply)
+        except BrokenPipeError:
+            self.logger.error("client disconnected unexpectedly.")
+            return -1
+        self.logger.debug("reply keep alive request.")
+        return 0
+
+    def store_async(self, conn, data, cmd_id):
+        """Parse and store async event."""
+
+        self.logger.debug("parse and store async event.")
+        self_conn = self.conn_vals[conn.fileno()]
+        nvme_sgl = struct.unpack_from('<16B', data, 32)
+        nvme_sgl_desc_type = nvme_sgl[15] & 0xF0
+        nvme_sgl_desc_sub_type = nvme_sgl[15] & 0x0F
+        nvme_async_dword10 = struct.unpack_from('<I', data, 48)[0]
+        nvme_async_dword11 = struct.unpack_from('<I', data, 52)[0]
+        nvme_async_dword12 = struct.unpack_from('<I', data, 56)[0]
+        nvme_async_dword13 = struct.unpack_from('<I', data, 60)[0]
+        nvme_async_dword14 = struct.unpack_from('<I', data, 64)[0]
+        nvme_async_dword15 = struct.unpack_from('<I', data, 68)[0]
+
+        self_conn.recv_async = True
+        self_conn.async_cmd_id = cmd_id
+
+    def _state_notify_update(self, update, is_add_req):
+        """Notify and reply async event."""
+
+        for key in list(self.conn_vals.keys()):
+            if self.conn_vals[key].recv_async is True:
+                async_reply = CqeNVMe()
+                # async_event_type:0x2 async_event_info:0xf0 log_page_identifier:0x70
+                async_reply.dword0 = b'\x02\xf0\x70\x00'
+                async_reply.sq_head_ptr = \
+                    struct.pack('<H', self.conn_vals[key].sq_head_ptr)
+                async_reply.cmd_id = \
+                    struct.pack('<H', self.conn_vals[key].async_cmd_id)
+                reply = async_reply.compose_reply()
+                try:
+                    self.conn_vals[key].connection.sendall(reply)
+                except BrokenPipeError:
+                    self.logger.error("client disconnected unexpectedly.")
+                    return -1
+                self.logger.debug("notify and reply async request.")
+                self.conn_vals[key].recv_async = False
+                return 0
+
+    def handle_timeout(self):
+        """Handle connection timeout."""
+
+        # The timeout time determination here is not very accurate.
+        # and instead of locking, a snapshot(every 0.1s) is used.
+        while True:
+            for key in list(self.conn_vals.keys()):
+                if self.conn_vals[key].keep_alive_timeout != 0 and \
+                  time.time() - self.conn_vals[key].keep_alive_time >= \
+                    self.conn_vals[key].keep_alive_timeout / 1000:
+                    self.logger.debug(f"discover request from {self.conn_vals[key].connection} timeout.")
+                    self.selector.unregister(self.conn_vals[key].connection)
+                    self.conn_vals[key].connection.close()
+                    del self.conn_vals[key]
+            time.sleep(0.1)
+
+    def reply_fabric_request(self, conn, data, cmd_id):
+        """Reply fabric request."""
+
+        fabric_type = struct.unpack_from('<B', data, 12)[0]
+        handle_fabric = {
+            NVME_TCP_FCTYPE.CONNECT: self.reply_fc_cmd_connect,
+            NVME_TCP_FCTYPE.PROP_GET: self.reply_fc_cmd_prop_get,
+            NVME_TCP_FCTYPE.PROP_SET: self.reply_fc_cmd_prop_set
+        }
+        class UnknownFabricType(BaseException):
+            def __init__(self, fabric_type):
+                super().__init__(f"unsupported opcode: {fabric_type}")
+        try:
+            err = handle_fabric[fabric_type](conn, data, cmd_id)
+        except KeyError as e:
+            raise UnknownFabricType(fabric_type) from e
+        return err
+
+    def nvmeof_tcp_connection(self, conn, mask):
+        """Handle request."""
+
+        err = 0
+        self_conn = self.conn_vals[conn.fileno()]
+        try:
+            message = conn.recv(NVME_TCP_PDU_UNIT)
+            if message:
+                self_conn.recv_buffer += message
+            else:
+                return
+        except BlockingIOError:
+            self.logger.error("recived data failed.")
+
+        while True:
+            if len(self_conn.recv_buffer) < 8:
+                return
+            pdu = struct.unpack_from('<BBBBI', self_conn.recv_buffer, 0)
+            pdu_type = pdu[0]
+            psh_flag = pdu[1]
+            ph_len = pdu[2]
+            ph_off = pdu[3]
+            package_len = pdu[4]
+            if len(self_conn.recv_buffer) < package_len:
+                return
+
+            data = self_conn.recv_buffer[:package_len]
+            self_conn.recv_buffer = self_conn.recv_buffer[package_len:]
+
+            self_conn.sq_head_ptr += 1
+            if self_conn.sq_head_ptr > SQ_HEAD_MAX:
+                self_conn.sq_head_ptr = 1
+
+            if NVME_TCP_PDU(pdu_type) == NVME_TCP_PDU.ICREQ:
+                err = self.reply_initialize(conn)
+
+            elif NVME_TCP_PDU(pdu_type) == NVME_TCP_PDU.CMD:
+                CMD1 = struct.unpack_from('<BBH', data, 8)
+                opcode = CMD1[0]
+                reserved = CMD1[1]
+                cmd_id = CMD1[2]
+
+                handle_opcode = {
+                    NVME_TCP_OPC.FABRIC_TYPE: self.reply_fabric_request,
+                    NVME_TCP_OPC.GET_LOG_PAGE: self.reply_get_log_page,
+                    NVME_TCP_OPC.IDENTIFY: self.reply_identify,
+                    NVME_TCP_OPC.SET_FEATURES: self.reply_set_feature,
+                    NVME_TCP_OPC.GET_FEATURES: self.reply_get_feature,
+                    NVME_TCP_OPC.KEEP_ALIVE: self.reply_keep_alive,
+                    NVME_TCP_OPC.ASYNC_EVE_REQ: self.store_async
+                }
+                class UnknownNVMEOpcode(BaseException):
+                    def __init__(self, opcode):
+                        super().__init__(f"unsupported opcode: {opcode}")
+                try:
+                    err = handle_opcode[opcode](conn, data, cmd_id)
+                except KeyError as e:
+                    raise UnknownNVMEOpcode(opcode) from e
+            else:
+                self.logger.error("unsupported pdu type: {pdu_type}")
+
+            if err == -1 or self_conn.shutdown_now is True:
+                del self.conn_vals[conn.fileno()]
+                self.selector.unregister(conn)
+                self.logger.debug(f"discover request from {conn} finished.")
+                conn.close()
+                return
+
+    def nvmeof_accept(self, sock, mask):
+        """Accept connection."""
+
+        conn, addr = sock.accept()
+        self.logger.debug(f"accept connection {conn} from {addr}")
+        conn.setblocking(False)
+        if conn.fileno() not in self.conn_vals:
+            self.conn_vals[conn.fileno()] = Connection()
+            self.conn_vals[conn.fileno()].connection = conn
+            global GLOBAL_CNLID
+            global GLOBAL_GEN_CNT
+            lock.acquire()
+            self.conn_vals[conn.fileno()].cnlid = GLOBAL_CNLID
+            self.conn_vals[conn.fileno()].gen_cnt = GLOBAL_GEN_CNT
+            GLOBAL_CNLID += 1
+            GLOBAL_GEN_CNT += 1
+            lock.release()
+        self.selector.register(conn, selectors.EVENT_READ, \
+                               self.nvmeof_tcp_connection)
+
+    def start_service(self):
+        """Enable listening on the server side."""
+
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        sock.bind((self.discovery_addr, int(self.discovery_port)))
+        sock.listen(MAX_CONNECTION)
+        sock.setblocking(False)
+        self.selector.register(sock, selectors.EVENT_READ, self.nvmeof_accept)
+        self.logger.debug("waiting for connection...")
+        t = threading.Thread(target=self.handle_timeout)
+        t.start()
+
+        omap_state = OmapGatewayState(self.config)
+        local_state = LocalGatewayState()
+        gateway_state = GatewayStateHandler(self.config, local_state,
+                                            omap_state, self._state_notify_update)
+        gateway_state.start_update()
+
+        try:
+          while True:
+            events = self.selector.select()
+            for key, mask in events:
+                callback = key.data
+                callback(key.fileobj, mask)
+        except KeyboardInterrupt:
+          for key in self.conn_vals:
+            self.conn_vals[key].connection.close()
+          self.selector.close()
+          self.logger.debug("received a ctrl+C interrupt. exiting...")
+
+def main(args=None):
+    # Set up root logger
+    logging.basicConfig()
+    logger = logging.getLogger()
+    logger.setLevel(logging.DEBUG)
+
+    parser = argparse.ArgumentParser(prog="python3 -m control",
+                                     description="Discover NVMe gateways")
+    parser.add_argument(
+        "-c",
+        "--config",
+        default="ceph-nvmeof.conf",
+        type=str,
+        help="Path to config file",
+    )
+    args = parser.parse_args()
+
+    config = GatewayConfig(args.config)
+    discovery_service = DiscoveryService(config)
+    discovery_service.start_service()
+
+if __name__ == "__main__":
+    main()