-
Notifications
You must be signed in to change notification settings - Fork 49
/
Copy pathdiscovery.py
1192 lines (1026 loc) · 47.3 KB
/
discovery.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#
# Copyright (c) 2021 International Business Machines
# All rights reserved.
#
# SPDX-License-Identifier: LGPL-3.0-or-later
#
# Authors: [email protected]
#
import argparse
import json
from .config import GatewayConfig
from .state import GatewayState, LocalGatewayState, OmapGatewayState, GatewayStateHandler
from .utils import GatewayLogger
from .utils import GatewayUtilsCrypto
from typing import Dict
import socket
import threading
import time
import enum
import uuid
import struct
import selectors
import os
from dataclasses import dataclass, field
from ctypes import LittleEndianStructure, c_ubyte, c_uint8, c_uint16, c_uint32, c_uint64
# 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
class TRANSPORT_TYPES(enum.IntFlag):
RDMA = 0x1
FC = 0x2
TCP = 0x3
INTRA_HOST = 0xfe
# Address family types
class ADRFAM_TYPES(enum.IntFlag):
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
@dataclass
class Connection:
"""Data used multiple times in each connection."""
connection: socket.socket = None
allow_listeners: list = field(default_factory=list)
log_page: bytearray = field(default_factory=bytearray)
recv_buffer: bytearray = field(default_factory=bytearray)
nvmeof_connect_data_hostid: tuple = tuple((c_ubyte * 16)())
nvmeof_connect_data_cntlid: int = 0
nvmeof_connect_data_subnqn: tuple = tuple((c_ubyte * 256)())
nvmeof_connect_data_hostnqn: tuple = tuple((c_ubyte * 256)())
sq_head_ptr: int = 0
unsent_log_page_len: int = 0
# NVM ExpressTM Revision 1.4, page 47
# see Figure 78: Offset 14h: CC – Controller Configuration
property_configuration: tuple = tuple((c_ubyte * 8)())
shutdown_now: bool = False
controller_id: uuid = None
gen_cnt: int = 0
recv_async: bool = False
async_cmd_id: int = 0
keep_alive_time: float = 0.0
keep_alive_timeout: int = 0
class AutoSerializableStructure(LittleEndianStructure):
def __add__(self, other):
if isinstance(other, LittleEndianStructure):
return bytes(self) + bytes(other)
elif isinstance(other, bytes):
return bytes(self) + other
else:
raise ValueError("error message format.")
class Pdu(AutoSerializableStructure):
_fields_ = [
("type", c_uint8),
("specical_flag", c_uint8),
("header_length", c_uint8),
("data_offset", c_uint8),
("packet_length", c_uint32),
]
class ICResp(AutoSerializableStructure):
_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)
]
class CqeConnect(AutoSerializableStructure):
_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)
]
class CqePropertyGetSet(AutoSerializableStructure):
_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)
]
class NVMeTcpDataPdu(AutoSerializableStructure):
_fields_ = [
("cmd_id", c_uint16),
("transfer_tag", c_uint16),
("data_offset", c_uint32),
("data_length", c_uint32),
("reserved", c_uint32)
]
class NVMeIdentify(AutoSerializableStructure):
_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)
]
# for set feature, keep alive and async
class CqeNVMe(AutoSerializableStructure):
_fields_ = [
("dword0", c_uint32),
("dword1", c_uint32),
("sq_head_ptr", c_uint16),
("sq_id", c_uint16),
("cmd_id", c_uint16),
("status", c_uint16)
]
class NVMeGetLogPage(AutoSerializableStructure):
_fields_ = [
# generation counter
("genctr", c_uint64),
# number of records
("numrec", c_uint64),
# record format
("recfmt", c_uint16),
("reserved", c_ubyte * 1006)
]
class DiscoveryLogEntry(AutoSerializableStructure):
_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)
]
class DiscoveryService:
"""Implements discovery controller.
Response discover request from initiator, this must be called from within a "with" block.
Instance attributes:
version: Discovery controller version
config: Basic gateway parameters
logger: Logger instance to track discovery controller events
omap_name: OMAP object name
discovery_addr: Discovery controller addr which allows initiator send command
discovery_port: Discovery controller's listening port
"""
def __init__(self, config):
self.version = 1
self.config = config
self.lock = threading.Lock()
self.omap_state = OmapGatewayState(self.config, f"discovery-{socket.gethostname()}")
self.gw_logger_object = GatewayLogger(config)
self.logger = self.gw_logger_object.logger
gateway_group = self.config.get_with_default("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}")
self.discovery_addr = self.config.get_with_default("discovery", "addr", "0.0.0.0")
self.discovery_port = self.config.get_with_default("discovery", "port", "8009")
if not self.discovery_addr or not 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.sock = None
self.conn_vals = {}
self.connection_counter = 1
self.selector = selectors.DefaultSelector()
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
if self.omap_state:
self.omap_state.cleanup_omap()
self.omap_state = None
if self.selector:
with self.lock:
for key in self.conn_vals:
try:
self.selector.unregister(self.conn_vals[key].connection)
except Exception:
pass
try:
self.conn_vals[key].connection.close()
except Exception:
pass
self.conn_vals = {}
if self.sock:
try:
self.selector.unregister(self.sock)
except Exception:
pass
try:
self.sock.close()
except Exception:
pass
self.sock = None
try:
self.selector.close()
except Exception:
pass
self.selector = None
def _read_all(self) -> Dict[str, str]:
"""Reads OMAP and returns dict of all keys and values."""
omap_dict = self.omap_state.get_state()
return omap_dict
def _get_vals(self, omap_dict, prefix):
"""Read values from the OMAP dict."""
return [json.loads(val.decode('utf-8')) for (key, val) in omap_dict.items()
if key.startswith(prefix)]
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
try:
conn.sendall(pdu_reply + icresp_reply + bytes(112))
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) # noqa: F841
SIGL1 = struct.unpack_from('<QI4B', data, 32)
address = SIGL1[0] # noqa: F841
length = SIGL1[1] # noqa: F841
reserved3 = SIGL1[2] # noqa: F841
descriptor_type = SIGL1[5] # noqa: F841
CMD2 = struct.unpack_from('<HHHBBI', data, 48)
record_format = CMD2[0] # noqa: F841
queue_id = CMD2[1] # noqa: F841
submission_queue_size = CMD2[2] # noqa: F841
connect_attributes = CMD2[3] # noqa: F841
keep_alive_timeout = CMD2[5]
self_conn = self.conn_vals[conn.fileno()]
self_conn.keep_alive_time = time.time()
if keep_alive_timeout == 0:
keep_alive_timeout = 15000
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 = int(self_conn.controller_id)
connect_reply.sq_head_ptr = self_conn.sq_head_ptr
connect_reply.cmd_id = cmd_id
try:
conn.sendall(pdu_reply + connect_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) # noqa: F841
# property size = (attrib+1)x4, 0x1 means 8 bytes
nvmeof_prop_get_set_attrib = struct.unpack_from('<1B', data, 48)[0] # noqa: F841
nvmeof_prop_get_set_rsvd1 = struct.unpack_from('<3B', data, 49) # noqa: F841
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_configuration
elif NVME_CTL(nvmeof_prop_get_set_offset) == NVME_CTL.STATUS:
shutdown_notification = (self_conn.property_configuration[1] >> 6) & 0x3
if shutdown_notification == 0:
# check CC.EN bit
enabled = self_conn.property_configuration[0] & 0x1
if enabled != 0:
# 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
try:
conn.sendall(pdu_reply + property_get)
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) # noqa: F841
nvmeof_prop_get_set_attrib = struct.unpack_from('<1B', data, 48)[0] # noqa: F841
nvmeof_prop_get_set_rsvd1 = struct.unpack_from('<3B', data, 49) # noqa: F841
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_configuration = struct.unpack_from('<8B', data, 56)
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
try:
conn.sendall(pdu_reply + property_set)
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] # noqa: F841
nvme_rsvd1 = struct.unpack_from('<Q', data, 16)[0] # noqa: F841
nvme_mptr = struct.unpack_from('<Q', data, 24)[0] # noqa: F841
nvme_sgl = struct.unpack_from('<16B', data, 32)
nvme_sgl_desc_type = nvme_sgl[15] & 0xF0 # noqa: F841
nvme_sgl_desc_sub_type = nvme_sgl[15] & 0x0F # noqa: F841
nvme_identify_dword10 = struct.unpack_from('<I', data, 48)[0] # noqa: F841
nvme_identify_dword11 = struct.unpack_from('<I', data, 52)[0] # noqa: F841
nvme_identify_dword12 = struct.unpack_from('<I', data, 56)[0] # noqa: F841
nvme_identify_dword13 = struct.unpack_from('<I', data, 60)[0] # noqa: F841
nvme_identify_dword14 = struct.unpack_from('<I', data, 64)[0] # noqa: F841
nvme_identify_dword15 = struct.unpack_from('<I', data, 68)[0] # noqa: F841
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 = int(self_conn.controller_id)
# 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)
if len(self_conn.nvmeof_connect_data_subnqn) == 256:
for i in range(256):
identify_reply.subnqn[i] = self_conn.nvmeof_connect_data_subnqn[i]
else:
self.logger.error("error subnqn.")
return -1
try:
conn.sendall(pdu_reply + nvme_tcp_data_pdu + identify_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] # noqa: F841
nvme_rsvd1 = struct.unpack_from('<Q', data, 16)[0] # noqa: F841
nvme_mptr = struct.unpack_from('<Q', data, 24)[0] # noqa: F841
nvme_sgl = struct.unpack_from('<16B', data, 32)
nvme_sgl_desc_type = nvme_sgl[15] & 0xF0 # noqa: F841
nvme_sgl_desc_sub_type = nvme_sgl[15] & 0x0F # noqa: F841
# dword10 may include Feature Identifier:
# Asynchronous Event Configuration (0x0b) (Not currently used)
nvme_set_features_dword10 = struct.unpack_from('<I', data, 48)[0] # noqa: F841
nvme_set_features_dword11 = struct.unpack_from('<I', data, 52)[0] # noqa: F841
nvme_set_features_dword12 = struct.unpack_from('<I', data, 56)[0] # noqa: F841
nvme_set_features_dword13 = struct.unpack_from('<I', data, 60)[0] # noqa: F841
nvme_set_features_dword14 = struct.unpack_from('<I', data, 64)[0] # noqa: F841
nvme_set_features_dword15 = struct.unpack_from('<I', data, 68)[0] # noqa: F841
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
try:
conn.sendall(pdu_reply + set_feature_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] # noqa: F841
nvme_rsvd1 = struct.unpack_from('<Q', data, 16)[0] # noqa: F841
nvme_mptr = struct.unpack_from('<Q', data, 24)[0] # noqa: F841
nvme_sgl = struct.unpack_from('<16B', data, 32)
nvme_sgl_desc_type = nvme_sgl[15] & 0xF0 # noqa: F841
nvme_sgl_desc_sub_type = nvme_sgl[15] & 0x0F # noqa: F841
nvme_get_features_dword10 = struct.unpack_from('<I', data, 48)[0] # noqa: F841
nvme_get_features_dword11 = struct.unpack_from('<I', data, 52)[0] # noqa: F841
nvme_get_features_dword12 = struct.unpack_from('<I', data, 56)[0] # noqa: F841
nvme_get_features_dword13 = struct.unpack_from('<I', data, 60)[0] # noqa: F841
nvme_get_features_dword14 = struct.unpack_from('<I', data, 64)[0] # noqa: F841
nvme_get_features_dword15 = struct.unpack_from('<I', data, 68)[0] # noqa: F841
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
try:
conn.sendall(pdu_reply + get_feature_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, GatewayState.LISTENER_PREFIX)
hosts = self._get_vals(my_omap_dict, GatewayState.HOST_PREFIX)
if len(self_conn.nvmeof_connect_data_hostnqn) != 256:
self.logger.error("error hostnqn.")
return -1
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] # noqa: F841
nvme_rsvd1 = struct.unpack_from('<Q', data, 16)[0] # noqa: F841
nvme_mptr = struct.unpack_from('<Q', data, 24)[0] # noqa: F841
# https://nvmexpress.org/wp-content/uploads/NVM-Express-1_4-2019.06.10-Ratified.pdf
# The SGL Data Block descriptor, defined in Figure 114, describes a data block.
nvme_sgl = struct.unpack_from('<16B', data, 32)
nvme_sgl_desc_type = nvme_sgl[15] & 0xF0 # noqa: F841
nvme_sgl_desc_sub_type = nvme_sgl[15] & 0x0F # noqa: F841
nvme_sgl_len = nvme_sgl[8]
nvme_sgl_len += (nvme_sgl[9] << 8) + (nvme_sgl[10] << 16) + (nvme_sgl[11] << 24)
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_numdl = struct.unpack_from('<H', data, 50)[0]
nvme_get_logpage_numdh = struct.unpack_from('<H', data, 52)[0]
nvme_get_logpage_numd = (nvme_get_logpage_numdh << 16) + nvme_get_logpage_numdl
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] # noqa: F841
nvme_get_logpage_dword14 = struct.unpack_from('<I', data, 64)[0]
nvme_get_logpage_dword15 = struct.unpack_from('<I', data, 68)[0] # noqa: F841
get_logpage_lid = nvme_get_logpage_dword10 & 0xFF
get_logpage_lsp = (nvme_get_logpage_dword10 >> 8) & 0x1F # noqa: F841
get_logpage_lsi = nvme_get_logpage_dword11 >> 16 # noqa: F841
get_logpage_uid_idx = nvme_get_logpage_dword14 & 0x3F # noqa: F841
if get_logpage_lid != 0x70:
self.logger.error("request type error, not discovery request.")
return -1
if nvme_data_len != nvme_sgl_len:
self.logger.error(f"request data len error, {nvme_data_len=} != {nvme_sgl_len=}.")
return -1
# Filter listeners based on host access permissions
allow_listeners = self_conn.allow_listeners
if len(allow_listeners) == 0:
for host in hosts:
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] = nvme_get_log_page_reply
# log entries
log_entry_counter = 0
while log_entry_counter < len(allow_listeners):
log_entry = DiscoveryLogEntry()
log_entry.trtype = TRANSPORT_TYPES.TCP
log_adrfam = allow_listeners[log_entry_counter]["adrfam"]
adrfam = ADRFAM_TYPES[log_adrfam.lower()]
if adrfam is None:
self.logger.error(f"unsupported address family {log_adrfam}")
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
str_trsvcid = str(allow_listeners[log_entry_counter]["trsvcid"])
log_entry.trsvcid = (c_ubyte * 32)(*[c_ubyte(x) for x in str_trsvcid.encode()])
log_entry.trsvcid[len(str_trsvcid):] = \
[c_ubyte(0x20)] * (32 - len(str_trsvcid))
# NVM subsystem qualified name
entry = allow_listeners[log_entry_counter]["nqn"]
log_entry.subnqn = (c_ubyte * 256)(*[c_ubyte(x) for x in entry.encode()])
log_entry.subnqn[len(allow_listeners[log_entry_counter]["nqn"]):] = \
[c_ubyte(0x00)] * (256 - len(allow_listeners[log_entry_counter]["nqn"]))
# Transport address
entry = allow_listeners[log_entry_counter]["traddr"]
log_entry.traddr = (c_ubyte * 256)(*[c_ubyte(x) for x in entry.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)] = 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 <= 1024 and nvme_logpage_offset == 0:
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)
reply = pdu_reply + nvme_tcp_data_pdu + bytes(nvme_get_log_page_reply)[:nvme_data_len]
elif nvme_data_len % 1024 == 0:
# reply log pages
reply = pdu_reply + 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.allow_listeners = []
else:
self.logger.error(f"request log page: invalid length error {nvme_data_len=}")
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()]
nvme_sgl = struct.unpack_from('<16B', data, 32)
nvme_sgl_desc_type = nvme_sgl[15] & 0xF0 # noqa: F841
nvme_sgl_desc_sub_type = nvme_sgl[15] & 0x0F # noqa: F841
nvme_keep_alive_dword10 = struct.unpack_from('<I', data, 48)[0] # noqa: F841
nvme_keep_alive_dword11 = struct.unpack_from('<I', data, 52)[0] # noqa: F841
nvme_keep_alive_dword12 = struct.unpack_from('<I', data, 56)[0] # noqa: F841
nvme_keep_alive_dword13 = struct.unpack_from('<I', data, 60)[0] # noqa: F841
nvme_keep_alive_dword14 = struct.unpack_from('<I', data, 64)[0] # noqa: F841
nvme_keep_alive_dword15 = struct.unpack_from('<I', data, 68)[0] # noqa: F841
pdu_reply = Pdu()
pdu_reply.type = NVME_TCP_PDU.RSP
pdu_reply.header_length = 24
pdu_reply.packet_length = 24
# Cqe for keep alive
keep_alive_reply = CqeNVMe()
keep_alive_reply.sq_head_ptr = self_conn.sq_head_ptr
keep_alive_reply.cmd_id = cmd_id
try:
conn.sendall(pdu_reply + keep_alive_reply)
except BrokenPipeError:
self.logger.error("client disconnected unexpectedly.")
return -1
self.logger.debug("reply keep alive request.")
return 0
def reply_not_supported(self, conn, data, cmd_id):
"""Reply to not supported opcodes."""
self.logger.warning("handle not supported opcode.")
self_conn = self.conn_vals[conn.fileno()]
pdu_reply = Pdu()
pdu_reply.type = NVME_TCP_PDU.RSP
pdu_reply.header_length = 24
pdu_reply.packet_length = 24
# Cqe for not supported opcode reply
not_supported_reply = CqeNVMe()
not_supported_reply.sq_head_ptr = self_conn.sq_head_ptr
not_supported_reply.cmd_id = cmd_id
# Invalid Command Opcode: A reserved coded value or an unsupported value
# in the command opcode field.
not_supported_reply.status = 1
try:
conn.sendall(pdu_reply + not_supported_reply)
except BrokenPipeError:
self.logger.error("client disconnected unexpectedly.")
return -1
self.logger.warning("reply not supported opcode.")
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 # noqa: F841
nvme_sgl_desc_sub_type = nvme_sgl[15] & 0x0F # noqa: F841
nvme_async_dword10 = struct.unpack_from('<I', data, 48)[0] # noqa: F841
nvme_async_dword11 = struct.unpack_from('<I', data, 52)[0] # noqa: F841
nvme_async_dword12 = struct.unpack_from('<I', data, 56)[0] # noqa: F841
nvme_async_dword13 = struct.unpack_from('<I', data, 60)[0] # noqa: F841
nvme_async_dword14 = struct.unpack_from('<I', data, 64)[0] # noqa: F841
nvme_async_dword15 = struct.unpack_from('<I', data, 68)[0] # noqa: F841
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."""
should_send_async_event = False
for key in update.keys():
if key.startswith(GatewayState.SUBSYSTEM_PREFIX):
should_send_async_event = True
break
if key.startswith(GatewayState.LISTENER_PREFIX):
should_send_async_event = True
break
if not should_send_async_event:
return
for key in list(self.conn_vals.keys()):
if self.conn_vals[key].recv_async is True:
pdu_reply = Pdu()
pdu_reply.type = NVME_TCP_PDU.RSP
pdu_reply.header_length = 24
pdu_reply.packet_length = 24
async_reply = CqeNVMe()
# async_event_type:0x2 async_event_info:0xf0 log_page_identifier:0x70
async_reply.dword0 = int.from_bytes(b'\x02\xf0\x70\x00', byteorder='little')
async_reply.sq_head_ptr = self.conn_vals[key].sq_head_ptr
async_reply.cmd_id = self.conn_vals[key].async_cmd_id
try:
self.conn_vals[key].connection.sendall(pdu_reply + async_reply)
except BrokenPipeError:
self.logger.error("client disconnected unexpectedly.")
return
self.logger.debug("notify and reply async request.")
self.conn_vals[key].recv_async = False
return
def handle_timeout(self):
"""Handle connection timeout."""
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:
# Adding locks to prevent another thread from processing sudden requests.
# Is there a better way?
with self.lock:
self.logger.debug(f"discover request from "
f"{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(1)
def reply_fabric_request(self, conn, data, cmd_id):
"""Reply fabric request."""
fabric_type = struct.unpack_from('<B', data, 12)[0]