ipfdemo_example.c Source File

Reference Documentation
Platform
Napatech SmartNIC
Content Type
Reference Information
Capture Software Version
Link™ Capture Software 12.15
Napatech Software Suite: examples/net/ipfdemo/ipfdemo_example.c Source File
ipfdemo_example.c
Go to the documentation of this file.
1 /*
2  *
3  * Copyright 2025 Napatech A/S. All Rights Reserved.
4  *
5  * 1. Copying, modification, and distribution of this file, or executable
6  * versions of this file, is governed by the terms of the Napatech Software
7  * license agreement under which this file was made available. If you do not
8  * agree to the terms of the license do not install, copy, access or
9  * otherwise use this file.
10  *
11  * 2. Under the Napatech Software license agreement you are granted a
12  * limited, non-exclusive, non-assignable, copyright license to copy, modify
13  * and distribute this file in conjunction with Napatech SmartNIC's and
14  * similar hardware manufactured or supplied by Napatech A/S.
15  *
16  * 3. The full Napatech Software License Agreement is included in this
17  * distribution, please see "NA-0009 Software License Agreement.pdf"
18  *
19  * 4. Redistributions of source code must retain this copyright notice,
20  * list of conditions and the following disclaimer.
21  *
22  * THIS SOFTWARE IS PROVIDED "AS IS" WITHOUT ANY WARRANTIES, EXPRESS OR
23  * IMPLIED, AND NAPATECH DISCLAIMS ALL IMPLIED WARRANTIES INCLUDING ANY
24  * IMPLIED WARRANTY OF TITLE, MERCHANTABILITY, NONINFRINGEMENT, OR OF
25  * FITNESS FOR A PARTICULAR PURPOSE. TO THE EXTENT NOT PROHIBITED BY
26  * APPLICABLE LAW, IN NO EVENT SHALL NAPATECH BE LIABLE FOR PERSONAL INJURY,
27  * OR ANY INCIDENTAL, SPECIAL, INDIRECT OR CONSEQUENTIAL DAMAGES WHATSOEVER,
28  * INCLUDING, WITHOUT LIMITATION, DAMAGES FOR LOSS OF PROFITS, CORRUPTION OR
29  * LOSS OF DATA, FAILURE TO TRANSMIT OR RECEIVE ANY DATA OR INFORMATION,
30  * BUSINESS INTERRUPTION OR ANY OTHER COMMERCIAL DAMAGES OR LOSSES, ARISING
31  * OUT OF OR RELATED TO YOUR USE OR INABILITY TO USE NAPATECH SOFTWARE OR
32  * SERVICES OR ANY THIRD PARTY SOFTWARE OR APPLICATIONS IN CONJUNCTION WITH
33  * THE NAPATECH SOFTWARE OR SERVICES, HOWEVER CAUSED, REGARDLESS OF THE THEORY
34  * OF LIABILITY (CONTRACT, TORT OR OTHERWISE) AND EVEN IF NAPATECH HAS BEEN
35  * ADVISED OF THE POSSIBILITY OF SUCH DAMAGES. SOME JURISDICTIONS DO NOT ALLOW
36  * THE EXCLUSION OR LIMITATION OF LIABILITY FOR PERSONAL INJURY, OR OF
37  * INCIDENTAL OR CONSEQUENTIAL DAMAGES, SO THIS LIMITATION MAY NOT APPLY TO YOU.
38  *
39  *
40 
41  */
42 
43 /**
44  * @example net/ipfdemo/ipfdemo_example.c
45  * @section ipfdemo_example_description Description
46  *
47  * This source file is an example of how to write an IP fragment re-assembling code,
48  * using the Napatech FPGA IPFMode feature. It accelerates IP re-assembling by
49  * making it possible to load balance using a 5 tuple hash algoritm instead of
50  * a 2 tuple hash algorithm. This is done for fragments by learning from the first
51  * fragment and use this knowledge to calculate the 5 tuple for belonging fragments.
52  * The complexing case is the situation where the FPGA was not able to learn
53  * and identify a fragment and therefore it is sent to one of the un-matching streams.
54  * In this situation the code needs to exchange the un-matched packet and deliver
55  * it to the right re-assembling thread.
56  *
57  *
58  * The following NTAPI functions are used:
59  * - @ref NT_Init()
60  * - @ref NT_NetRxOpen()
61  * - @ref NT_NTPL()
62  * - @ref NT_NetRxGet()
63  * - @ref NT_NET_GET_PKT_L2_FRAME_TYPE()
64  * - @ref NT_NET_GET_PKT_L3_FRAME_TYPE()
65  * - @ref NT_NET_GET_PKT_L3_FRAGMENTED()
66  * - @ref NT_NET_GET_PKT_DESCRIPTOR_FORMAT()
67  * - @ref NT_NET_GET_PKT_L3_FIRST_FRAG()
68  * - @ref NT_NET_GET_PKT_L2_PTR()
69  * - @ref NT_NET_GET_PKT_L3_OFFSET()
70  * - @ref NT_NET_GET_SEGMENT_TIMESTAMP()
71  * - @ref NT_NET_GET_PKT_IPF_LAST_FRAGMENT()
72  * - @ref NT_NET_GET_PKT_IPF_UNMATCHED_STREAMID()
73  * - @ref NT_NetRxRelease()
74  * - @ref NT_NetRxClose()
75  * - @ref NT_ExplainError()
76  *
77  * @section ipfdemo_example_prerequisites Prerequisites
78  * A Napatech capture accelerator is needed to run this example with IPFMode support.
79  * The ntservice.ini file must have enough HostBuffersRx defined to honor
80  * the requested streams needed when running this demo application.
81  * Below is an example of a minimum ini-file. It will create 20 64MB RX
82  * hostbuffer from NUMA node 0.
83  *
84  * Note that the PacketDescriptor is set to Ext9, This is needed to run
85  * IPFMode at all. If anything less is specified, then the IPFMode will
86  * not be used and it will revert into a Hash2Tuple mode.
87  *
88  * Note2 that the TimeSyncReferencePriority is set to OsTime. This is critical, when
89  * using a fragtimeout value (default), because the timestamp of each
90  * packet is used to calculate its timeout.
91  * @code
92  * [System]
93  * TimestampFormat = NATIVE_UNIX
94  *
95  * [Adapter0]
96  * AdapterType = NT20E2
97  * BusId = 00:02:00:00
98  * PacketDescriptor = Ext9
99  * HostBuffersRx = [20,64,0]
100  * TimeSyncReferencePriority = OsTime
101  * @endcode
102  *
103  *
104  * ************************* Overview of the algorithm used ******************************
105  *
106  * The FPGA enables you to match IP fragments and distribute them using a multi CPU
107  * buffer splitting scheme based on a 5-tuple hash algorithm when using the IPFMode feature.
108  * If a multi CPU splitting scheme based on a 2-tuple hash algorithm gives good enough
109  * distribution, then the IPFMode feature in the FPGA should not be used. All belonging
110  * fragments will end up in same stream using a 2-tuple hash algorithm.
111  * IPFMode: how it works:
112  * - Learns from first fragment
113  * - First fragment contains streamID of unmatched fragments
114  * - Timeout setup for releasing learned entry
115  *
116  * @code
117  * Graphical overview:
118  *
119  * +--------------+
120  * | DOI hash |
121  * +------>| |------+
122  * | | tables | |
123  * | +--------------+ |
124  * | |
125  * | |
126  * +----------------+ | | +----------------+
127  * | |------+ +----->| |
128  * | Reassembling | +--------------+ | Unmatched |
129  * | | | Return | | |
130  * | threads |------------->| msg box |------------>| threads |
131  * | | | FIFOs | | |
132  * | |<-----+ +--------------+ +------| |
133  * +----------------+ | | +----------------+
134  * ^ ^ ^ | | ^ ^
135  * | | | | | | |
136  * | | | | | | |
137  * | | | | +--------------+ | | |
138  * | | | | | Msg box | | | |
139  * | | | +-------| |<-----+ | |
140  * | | | | FIFOs | | |
141  * | | | +--------------+ | |
142  * | | | | |
143  * 5 tuple hash splitting 2 tuple hash splitting
144  * | | | | |
145  * | | | | |
146  * +-------------------------------------------------------------------------------+
147  * | |
148  * | NTAPI/FPGA |
149  * | |
150  * +-------------------------------------------------------------------------------+
151  *
152  * @endcode
153  *
154  * Description of algorithm:
155  *
156  *
157  * Threads:
158  * --------
159  *
160  * Unmatched threads – variable number (N) – number of unmatched streams to use
161  * (2-tuple hash cpu-splitting)
162  *
163  * Each unmatched thread receives the unmatched fragments. When a fragment is received,
164  * it is checked in the DOI tables to find a datagram ID match. If a match is found, the
165  * NetBuf containing the fragment is send to that reassembling Msg box FIFO, of which
166  * the DOI table belongs to. Otherwise, if no match found (yet), the fragment is
167  * temporarily put into a wait list. The wait list is periodically checked against the
168  * DOI tables.
169  *
170  * Reassembling threads – variable number (M) – number of streams to use
171  * (5-tuple hash cpu-splitting)
172  *
173  * When each reassembling thread receives IP fragments, it collects them into complete
174  * datagrams. If all fragments are received in order, or at least the first fragment is
175  * received first, then the need for unmatched fragment collection is not needed and all
176  * datagrams are reassembled in the reassembling threads.
177  * When the reassembling thread receives a first fragment, it then stores that fragment
178  * in its local collection hash table (tbl) and makes that datagram ID available for the
179  * specific unmatched thread to inform about its interest in fragments belonging to this
180  * datagram, using the specific DOI table connecting the reassembling thread with the
181  * unmatched thread.
182  * When all fragments are received for a datagram, they are all released and the DOI
183  * table entry is cleared.
184  * Each reassembling thread first reads from the Msg box for a NetBuf, and if none found,
185  * then it reads the NTAPI stream for a NetBuf. If any fragments are received from here,
186  * they are put into the collection table (tbl).
187  *
188  *
189  * Inter-thread communication messages:
190  * ------------------------------------
191  *
192  * Msg boxes (FIFOs):
193  *
194  * Used by Unmatched threads to send unmatched fragments to the reassembling threads.
195  * [[[unm1 Msgbox],[ unm2 Msgbox],…[ unmN Msgbox]], [[unm1 Msgbox],[ unm2 Msgbox],…[ unmN Msgbox]]…M times]
196  * reasm1 reasm2 …reasmM
197  *
198  * The Msg boxes are built as FIFOs and are used by the unmatched threads to send the
199  * NetBuf of a received datagram fragment to the belonging reassembling thread.
200 
201  * Return Msg boxes (FIFOs):
202  *
203  * Used by Reassmbling threads to send unmatched-fragment NetBufs back to the unmatched
204  * threads originally received them.
205  * [[[reasm1 Msgbox],[ reasm2 Msgbox],…[ reasmM Msgbox]], [[reasm1 Msgbox],[ reasm2 Msgbox],…[ reasmN Msgbox]]…N times]
206  * unm1 unm2 …unmN
207  *
208  * This FIFO is only needed to make the complete algorithm lockless. The NTAPI cannot
209  * handle multi-threaded access to the packet interface without serialization.
210 
211  * DOI tables (Datagram of interest):
212  *
213  * The DOI tables are a set of hash tables containing a list of datagram IDs that the
214  * reassembling threads are interested in.
215  *
216  * The FPGA specifies the unmatched stream ID together with the reception of the first
217  * fragment. This information is used by the reassembling threads to inform the specific
218  * unmatched thread about where to send unmatched fragments matching the first fragments
219  * datagram ID.
220  *
221  * Each reassembling thread has one DOI table allocated for each unmatched thread, thus
222  * M*N DOI hash tables are used. This way each reassembling thread has exclusive write
223  * access to one table for each unmatched tread.
224  * [[[reasm1 tbl],[reasm2 tbl],…[reasmM tbl]], [[reasm1 tbl],[reasm2 tbl],…[reasmM tbl]]…N times]
225  * Unmatched1 Unmatched2 …UnmatchedN
226  *
227  * Each unmatched thread only reads information from the DOI tables. When an unmatched
228  * fragment matches an entry in the DOI table, then this fragment is send to the
229  * reassembling thread using the dedicated Msg box FIFO.
230  *
231  * Datagram ID calculation:
232  *
233  * The fragments belonging to a datagram are identified by the calculated datagram ID:
234  * It consists of information from 4 fields in the IP header (source IP, destination IP,
235  * Identification and protocol).
236 
237  *<hr>
238  * @section ipfdemo_example_code Code
239  * @}
240  */
241 
242 // Include this in order to access the Napatech API
243 #include <nt.h>
244 
245 #if defined(__linux__) || defined(__FreeBSD__)
246  #include <unistd.h>
247  #include <signal.h>
248  #include <assert.h>
249  #include <pthread.h>
250  #include <arpa/inet.h>
251  #include <sys/time.h>
252  #include <stdatomic.h>
253 #elif defined(WIN32) || defined (WIN64)
254  #include <winsock2.h> // ntohs()
255  #include <time.h>
256  #include <sys/timeb.h>
257  #include <process.h> // threading
258 #endif
259 
260 #include <stdlib.h>
261 #include <argparse.h>
262 
263 #if defined(WIN32) || defined (WIN64)
264  #define snprintf(dst, ...) _snprintf_s((dst), _countof(dst), __VA_ARGS__)
265 
266 #if defined(_MSC_VER)
267  // Time
268  struct timezone {
269  int tz_minuteswest; /* minutes west of Greenwich */
270  int tz_dsttime; /* type of DST correction */
271  };
272 #endif // _MSC_VER
273 
274  int gettimeofday(struct timeval *tv, struct timezone *tz);
275 
276  void sleep(int time);
277  void usleep(unsigned long usec);
278 
279  // Threading
280  typedef HANDLE pthread_t;
281  typedef unsigned (__stdcall *start_address_t)(void *parameter);
282  int pthread_create(HANDLE *thread, DWORD *attr, start_address_t start_routine, void *parameter);
283  int pthread_join(HANDLE thread, void **value_ptr);
284 
285 #endif
286 
287 #if defined(WIN32) || defined (WIN64)
288 volatile int appRunning = 0; // The application will run as long as appRunning == 1
289 #else
290 atomic_int appRunning = 0; // The application will run as long as appRunning == 1
291 #endif
292 
293 #pragma pack(push, 1)
294 /* IPv4 header structure */
295 typedef struct iphdr_s {
296  uint8_t ipHlen:4; // little endian layout
297  uint8_t ipVer:4; // little endian layout
298  uint8_t tos;
299  uint16_t ipTotlen;
300  uint16_t Id;
301 #define FRAG_OFFS_MASK 0x1FFF
302 #define LAST_FRAG_BITS 0x00E0
303  uint16_t fragOffs;
304  uint8_t ttl;
305  uint8_t prot;
306  uint16_t csum;
307  uint32_t srcAddr;
308  uint32_t dstAddr;
309 } iphdr_t;
310 #pragma pack(pop)
311 
312 #define L3_ADDR(_NetBuf_) ((uint8_t *)NT_NET_GET_PKT_L2_PTR(_NetBuf_)+NT_NET_GET_PKT_L3_OFFSET(_NetBuf_))
313 #define IPVERSION(_NetBuf_) (((iphdr_t *)L3_ADDR(_NetBuf_))->ipVer)
314 
315 /* macros for IPv4 packets */
316 #define IPV4_HDR_LENGTH(_NetBuf_) ((uint8_t)(((iphdr_t *)L3_ADDR(_NetBuf_))->ipHlen)<<2)
317 #define IPV4_TOT_LEN(_NetBuf_) (ntohs(((iphdr_t *)L3_ADDR(_NetBuf_))->ipTotlen))
318 #define IPV4_FRAGMENT_ID(_NetBuf_) (ntohs(((iphdr_t *)L3_ADDR(_NetBuf_))->Id))
319 #define IPV4_FRAGMENT_OFFSET(_NetBuf_) ((ntohs(((iphdr_t *)L3_ADDR(_NetBuf_))->fragOffs)&FRAG_OFFS_MASK)<<3)
320 #define IPV4_LAST_FRAG(_NetBuf_) (((((iphdr_t *)L3_ADDR(_NetBuf_))->fragOffs)&LAST_FRAG_BITS)==0)
321 #define IPV4_SRC_ADDR(_NetBuf_) (((iphdr_t *)L3_ADDR(_NetBuf_))->srcAddr)
322 #define IPV4_DST_ADDR(_NetBuf_) (((iphdr_t *)L3_ADDR(_NetBuf_))->dstAddr)
323 #define IPV4_PROTOCOL(_NetBuf_) (((iphdr_t *)L3_ADDR(_NetBuf_))->prot)
324 
325 #define IPV4_GET_DGRAM_ID(_NetBuf_, _src_, _dst_, _id_, _prot_) \
326  do {iphdr_t *_ip_=(iphdr_t *)L3_ADDR(_NetBuf_);_src_=_ip_->srcAddr;_dst_=_ip_->dstAddr;\
327  _id_=_ip_->Id;_prot_=_ip_->prot;}while(0)
328 
329 #define IPV4_DATA_LEN(_NetBuf_) (IPV4_TOT_LEN(_NetBuf_)-IPV4_HDR_LENGTH(_NetBuf_))
330 
331 
332 /* Help structure to concatenate ID source fields */
333 typedef union _DOI_dgramId_u {
334  volatile uint64_t src_id;
335  struct {
336  volatile uint32_t src;
337  volatile uint32_t id;
338  };
339 } DOI_dgramId_u;
340 
341 #define MAX_SRC_ID 8
342 typedef struct _DOI_dgramTbl_t {
343  volatile uint64_t aSrc_id[MAX_SRC_ID]; // Source and frag ID concatenated into ID1
344  volatile uint64_t aDst_pr[MAX_SRC_ID]; // Destination and protocol concatenated into ID2
345 } DOI_dgramTbl_t;
346 
347 #define DOI_FRAG_TBL_SIZE 8081 /* Prime number */
348 #define DOI_HASH_GET_KEY(_entry_, _src_, _id_) (_entry_ = (_src_+_id_)%DOI_FRAG_TBL_SIZE)
349 
350 
351 /* Re-assembler hash table types and definitions */
352 struct frag_s {
353  NtNetBuf_t hNetBuf; // hNetBuf containing fragment buffer
354  uint16_t offset; // Fragment offset
355  uint16_t size; // Size of this fragment
356  uint8_t firstFrag; // If marked as first fragment
357  uint8_t lastFrag; // If marked as last fragment
358  uint8_t fromUnm; // Is this packet received from an un-matched thread (needed to lock on release)
359  uint8_t unmIndex; // if from unmatched stream, then this tells which unmatched thread index
360 };
361 
362 #define MAX_FRAG_CNT 18
363 #define REASSEMBLY_HASH_TBL_SIZE 1021 /* Prime number */
364 #define INITIAL_HASH_TBL_ENTRY_CNT 1024
365 typedef struct _hashTbl_entry_s {
366  uint64_t id1; // Src and prot id for ID1
367  uint64_t id2; // Dest and prot Id for ID2
368  volatile uint64_t *pDOI_Src_id; // Source and frag ID field
369  volatile uint64_t *pDOI_Dst_pr; // Dest and protocol ID field
370  uint16_t fragCnt; // Number of fragments in aFrag
371  struct frag_s aFrag[MAX_FRAG_CNT]; // Array of all fragments belonging to the same datagram
372  struct _hashTbl_entry_s *pNext; // Pointer to next element in list
373 } tbl_entry_t;
374 
375 /* Hash table macros */
376 #define LOOKUP_ENTRY(_tbl_base_, _tbl_entry_type_, _tbl_size_, _id1_, _id2_, _tbl_entry_) \
377 { \
378  uint32_t _key_ = (uint32_t)((_id1_ ^ _id2_) % _tbl_size_); \
379  _tbl_entry_type_ *_tbl_; \
380  _tbl_ = _tbl_base_[_key_]; \
381  while (_tbl_ && (_tbl_->id1 != _id1_ || _tbl_->id2 != _id2_)) { \
382  _tbl_ = _tbl_->pNext; \
383  } \
384  _tbl_entry_=_tbl_; \
385 }
386 
387 #define GET_NEW_TBL_ENTRY(_tbl_free_, _tbl_entry_type_, _tbl_entry_) \
388 { \
389  if (!_tbl_free_) { \
390  int __o__; \
391  _tbl_entry_type_ *__plm__; \
392  for (__o__ = 0; __o__ < 64; __o__++) { \
393  __plm__ = calloc(1, sizeof(_tbl_entry_type_)); \
394  if (!__plm__) assert(0); \
395  __plm__->pNext = _tbl_free_; \
396  _tbl_free_ = __plm__; \
397  } \
398  } \
399  _tbl_entry_ = _tbl_free_; \
400  _tbl_free_ = _tbl_free_->pNext; \
401  tbl_entry->pNext = NULL; \
402 }
403 
404 #define RELEASE_TBL_ENTRY(_tbl_free_, _tbl_entry_) \
405 { \
406  _tbl_entry_->pNext = _tbl_free_; \
407  _tbl_free_ = _tbl_entry_; \
408 }
409 
410 #define ADD_ENTRY_TO_TBL(_tbl_base_, _tbl_entry_type_, _tbl_size_, _tbl_entry_) \
411 { \
412  uint32_t _key_ = (uint32_t)((_tbl_entry_->id1 ^ _tbl_entry_->id2) % _tbl_size_); \
413  _tbl_entry_type_ *_tbl_; \
414  _tbl_ = _tbl_base_[_key_]; \
415  while (_tbl_ && (_tbl_->id1 != _tbl_entry_->id1 || _tbl_->id2 != _tbl_entry_->id2)) { \
416  _tbl_ = _tbl_->pNext; \
417  } \
418  if (_tbl_ == NULL) { \
419  _tbl_entry_->pNext = _tbl_base_[_key_]; \
420  _tbl_base_[_key_] = _tbl_entry_; \
421  } \
422  else { \
423  assert(_tbl_==_tbl_entry_); \
424  } \
425 }
426 
427 #define DEL_ENTRY_FROM_TBL(_tbl_base_, _tbl_entry_type_, _tbl_size_, _tbl_entry_) \
428 { \
429  uint32_t _key_ = (uint32_t)((_tbl_entry_->id1 ^ _tbl_entry_->id2) % _tbl_size_); \
430  _tbl_entry_type_ *_tbl_, *_prev_ = NULL; \
431  _tbl_ = _tbl_base_[_key_]; \
432  while (_tbl_ && (_tbl_->id1 != _tbl_entry_->id1 || _tbl_->id2 != _tbl_entry_->id2)) { \
433  _prev_ = _tbl_;_tbl_ = _tbl_->pNext; \
434  } \
435  if (_tbl_) { \
436  if (_prev_) { \
437  _prev_->pNext = _tbl_->pNext; \
438  } \
439  else { \
440  _tbl_base_[_key_] = _tbl_->pNext; \
441  } \
442  } \
443  else { \
444  assert(0); \
445  } \
446 }
447 
448 
449 typedef struct _ipDefrag ipDefrag_t;
450 
451 
452 #if defined(__linux__) || defined(__FreeBSD__)
453  #define compiler_barrier() __asm__ __volatile__("":::"memory")
454 #elif defined(WIN32) || defined (WIN64)
455  #define compiler_barrier()
456 #endif
457 
458 
459 /* Msg box definition and control type and macros */
460 #define MSG_BOX_DEPTH 0x2000
461 typedef struct _msg_box_s {
462  volatile uint32_t rdIdx; // Read index of msg box
463  volatile uint32_t wrIdx; // Write index of msg box
464  NtNetBuf_t data[MSG_BOX_DEPTH]; // Msg box elements (pointers to hNetBufs)
465 } msg_box_t;
466 
467 /* Macros to use the msg box */
468 #define MSG_BOX_EMPTY(_fifo_) ((_fifo_)->wrIdx==(_fifo_)->rdIdx)
469 #define MSG_BOX_FULL(_fifo_) (((_fifo_)->wrIdx-(_fifo_)->rdIdx) >= MSG_BOX_DEPTH)
470 #define MSG_BOX_GET(_fifo_) (_fifo_)->data[(_fifo_)->rdIdx&(MSG_BOX_DEPTH-1)]; compiler_barrier(); (_fifo_)->rdIdx++
471 #define MSG_BOX_PUT(_fifo_, _elem_) \
472  do { \
473  (_fifo_)->data[(_fifo_)->wrIdx&(MSG_BOX_DEPTH-1)]=_elem_; \
474  compiler_barrier(); \
475  (_fifo_)->wrIdx++; \
476  } while (0)
477 
478 
479 
480 /* Wait-list type definition */
481 #define MAX_WAIT_ELEM 1024
482 typedef struct wait_list_s {
483  NtNetBuf_t hNetBuf;
484  struct wait_list_s *pNext;
485 } wait_list_t;
486 
487 /* Un-matched thread variable definition */
488 typedef struct _unmThread_s {
489  pthread_t thread; // This thread
490  uint32_t streamId; // NT Stream Id to service by this thread
491  DOI_dgramTbl_t *pDOI; // Datagram Of Interest tables. For this Unmatched Thread only, indexed for each Reassembly Thread
492  msg_box_t *pMsgboxReturn; // All msg boxes used by the current unmatched thread indexed [reasmCnt]
493  wait_list_t WaitListElem[MAX_WAIT_ELEM]; // All awailable wait-list elements
494  wait_list_t *pFree; // Linked list of free wait-list elements
495  wait_list_t *pWait; // Linked list of waiting wait-list elements
496  ipDefrag_t *pIpDefrag; // Pointer to main instance variable
497  /* Statistics */
498  int fragRecv; // Counter for all fragments received by this un-matched thread
499  int fragsDeleted; // Counter for all timed out and deleted fragments in wait-list
500 } unmThread_t;
501 
502 
503 /* Re-assembling thread variable definition */
504 typedef struct _reasmThread_s {
505  pthread_t thread; // This thread
506  uint32_t idx; // Index number of this re-assembling thread (first re-assembling thread starts with index 0)
507  uint32_t streamId; // NT Stream Id to service by this thread
508  tbl_entry_t *tbl[REASSEMBLY_HASH_TBL_SIZE]; // Hash table for the current re-assembling thread
509  tbl_entry_t *tblFree; // Free elements for the hash table
510  msg_box_t *pMsgbox; // All msg boxes used by the current re-assembling thread indexed [unmCnt]
511  ipDefrag_t *pIpDefrag; // Pointer to main instance variable
512  /* statistics */
513  int datagramCompleted; // Counter for datagrams sent in fragments and re-assembled sucessfully
514  int fragmentsTimedout; // Counter for all timed out and deleted fragments in hash table
515  int firstFragRcv; // Counter for all first-fragments received by this re-assembling thread
516  int nonFragments; // Counter for all non-fragmented packets received by this re-assembling stream
517  int Src_IdClash; // Counter for all Id clashes encountered (frags with src,dst,prot,fragId identical)
518  int msgboxPackets; // Counter for remaining msg box entries on exit
519 } reasmThread_t;
520 
521 struct _ipDefrag {
522  int adapterNo; // Adapter number to run on
523  uint32_t unmCnt; // number of un-matched streams
524  uint32_t unmStart; // Value of first un-matched streamid
525  uint32_t reasmCnt; // Number of re-assembling streams
526  uint32_t reasmStart; // Value of first re-assembling streamId
527  uint64_t fragTimeout; // Fragment timeout setting in ms
528  unmThread_t *pUnmThreads; // All un-matched thread instance pointers
529  DOI_dgramTbl_t *pDOI_Tbls; // Hash tables to notify interest of fragments to the Unmatched Threads - indexed [unmCnt][reasmCnt]
530  reasmThread_t *pReasmThreads; // All re-assembling thread instance pointers
531  msg_box_t *pReasmMsgboxes; // Pointer to all re-assembling threads message boxes
532  msg_box_t *pReasmReturnMsgboxes; // Pointer to all NetBuf return message boxes
533  uint32_t extDescrType; // The NT extended descriptor type used by FPGA to access and use the correct macros
534  int allReasmClosed; // To let all Unm-threads/streams close after re-assembling threads are closed
535 };
536 
537 
538 #define NO_OPTIONS 0
539 #define OPTION_HELP (1<<1)
540 #define OPTION_ADAPTER (1<<2)
541 #define OPTION_NUM_REASM (1<<3)
542 #define OPTION_NUM_UNM (1<<4)
543 #define OPTION_TIMEOUT (1<<5)
544 #define OPTION_TABLE_PERSIST (1<<6)
545 #define OPTION_TABLE_TIMEOUT (1<<7)
546 
547 static int opt_adapter = -1;
548 static int opt_reasm = -1;
549 static int opt_unm = -1;
550 static int opt_frag = -1;
551 static char *opt_persist = NULL;
552 static int opt_timeout = -1;
553 
554 /**
555  * Table of valid options.
556  */
557 struct argparse_option arg_options[] = {
558  OPT_HELP(),
559  OPT_INTEGER('a', "adapter", &opt_adapter, "The adapter to run tests on", NULL, 0, 0, "adapter number"),
560  OPT_INTEGER('r', "reasm", &opt_reasm, "Number of concurrent IP fragments re-assembling streams/threads", NULL, 0, 0, "number"),
561  OPT_INTEGER('u', "unm", &opt_unm, "Number of concurrent un-matched streams/threads", NULL, 0, 0, "number"),
562  OPT_INTEGER('f', "fragtimeout", &opt_frag, "How old fragments may get before they are deleted from tables (ms)", NULL, 0, 0, "ms"),
563  OPT_STRING( 'p', "tablepersist", &opt_persist, "FPGA TablePersist", NULL, 0, 0, "timeout|lastfragment"),
564  OPT_INTEGER('t', "tabletimeout", &opt_timeout, "FPGA table timeout", NULL, 0, 0, "value"),
565  OPT_END(),
566 };
567 
568 static struct _ipDefrag IpDefrag; // The main instance handle
569 
570 static const char *usageText[] = {
571  "Syntax:\n"
572  "\n"
573  "ipfdemo_example [-a <adapter number>] [-r <number>] [-u <number>] [-f <ms>] [-p <timeout|lastfragment>] [-t <value>]\n"
574  "\nCommands:\n",
575  NULL};
576 
577 /*****************************************************************************
578  Function to get system time in milliseconds
579 ******************************************************************************/
580 static uint64_t GetSystemTimeNoArg(void)
581 {
582  struct timeval tv;
583  gettimeofday(&tv, NULL);
584  return (uint64_t) (tv.tv_sec * 1000000 + tv.tv_usec);
585 }
586 
587 /*****************************************************************************
588  Called from the un-matched threads, to check if any re-assembling threads
589  are waiting for this fragment. It is communicated using the DOI hash table.
590  One table for each re-assembling/un-matched thread combination. This is done
591  to make singular write access to each table (no locks needed).
592 ******************************************************************************/
593 static int SendToReasm(unmThread_t *pUnm, NtNetBuf_t *phNetBuf)
594 {
595  unsigned i;
596  uint32_t entry;
597  uint32_t src, dst;
598  uint8_t prot;
599  uint16_t id;
600  DOI_dgramId_u val, val1;
601  DOI_dgramTbl_t *pTbl;
602  msg_box_t *pMsgbox;
603  int idx;
604  NtNetBuf_t hNetBuf = *phNetBuf;
605 
606  IPV4_GET_DGRAM_ID(hNetBuf, src, dst, id, prot);
607 
608  val.src = src;
609  val.id = id;
610  val1.src = dst;
611  val1.id = prot;
612 
613  for (i = 0; i < pUnm->pIpDefrag->reasmCnt; i++) {
614  pTbl = &pUnm->pDOI[i * DOI_FRAG_TBL_SIZE];
615  DOI_HASH_GET_KEY(entry, src, id);
616  pMsgbox = &pUnm->pIpDefrag->pReasmMsgboxes[i * pUnm->pIpDefrag->unmCnt + (pUnm->streamId - pUnm->pIpDefrag->unmStart)];
617 
618  idx=0;
619  while (idx < MAX_SRC_ID) {
620  if (pTbl[entry].aSrc_id[idx] == val.src_id &&
621  pTbl[entry].aDst_pr[idx] == val1.src_id) {
622  if (!MSG_BOX_FULL(pMsgbox)) {
623  MSG_BOX_PUT(pMsgbox, hNetBuf);
624  *phNetBuf = NULL;
625  return 1;
626  } else {
627  /* Msgbox full, we try again later */
628  return 0;
629  }
630  }
631  idx++;
632  }
633  }
634  return 0;
635 }
636 
637 /*****************************************************************************
638  This is the main un-matched thread routine. One thread for each un-matched
639  stream is spawned to handle all the un-matched streams configured.
640  Upon receival of an IP fragment from hardware, this routine searches the DOI
641  tables (restricted to the ones associated with this stream ID) for interest
642  in this fragment from any re-assembling threads. If one is waiting for fragments
643  belonging to this particular datagram, it is send to that re-assembling thread
644  using the msg box mechanism. If none interested (yet), the fragment is temporarily
645  stored in a wait-list. Frequently the wait-list is scanned together with the DOI
646  tables to make sure the fragments are sent to the corresponding re-assembling
647  thread when it raises interest herein.
648  If the wait-list is full, no more fragments are read until the last hNetBuf
649  can be delivered or added to wait-list.
650 ******************************************************************************/
651 #if defined(WIN32) || defined (WIN64)
652  static unsigned __stdcall _UnMatchedThread(void *arg)
653 #else
654  static void *_UnMatchedThread(void *arg)
655 #endif
656 {
657  unmThread_t *pUnm = (unmThread_t *)arg; // this pointer for this thread
658  char streamName[20]; // Stream name
659  char errorBuffer[NT_ERRBUF_SIZE]; // Error buffer
660  NtNetStreamRx_t hNetRx; // Handle to the RX stream
661  NtNetBuf_t hNetBuf = NULL; // Net buffer container
662  unsigned i; // loop variable
663  int status; // status from NTAPI calls
664  int invalidPkt; // Invalid packet indicator (we only receive IP fragments with these streams)
665  int checkTimeout = 0; // Indicator for fragment timeout check
666  uint64_t sysTs = 0; // System Timestamp
667  int deleteIt; // If we should delete the entry from the wait-list. Is delivered or timed out
668  wait_list_t *pWait; // Help pointer to cleanup wait list on exit
669 
670  snprintf(streamName, sizeof(streamName), "IPFUnm_%i", pUnm->streamId);
671 
672  /* Fill free list with elements for the wait-list */
673  pUnm->WaitListElem[0].pNext = NULL;
674  for (i = 1; i < MAX_WAIT_ELEM; i++) {
675  pUnm->WaitListElem[i].pNext = &pUnm->WaitListElem[i - 1];
676  }
677  pUnm->pFree = &pUnm->WaitListElem[MAX_WAIT_ELEM - 1];
678 
679  // Get a stream handle
680  if ((status = NT_NetRxOpen(&hNetRx, streamName, NT_NET_INTERFACE_PACKET, pUnm->streamId, -1)) != NT_SUCCESS) {
681  // Get the status code as text
682  NT_ExplainError(status, errorBuffer, sizeof(errorBuffer));
683  fprintf(stderr, "[%i] NT_NetRxOpen() failed: %s\n", pUnm->streamId, errorBuffer);
684 #if defined(WIN32) || defined (WIN64)
685  { _endthreadex(0); return 0; }
686 #else
687  return (void *)NULL;
688 #endif
689  }
690 
691  while (appRunning == 1) {
692 
693  /* Check all Return Msg boxes for fragment packets to be freed */
694  for (i = 0; i < pUnm->pIpDefrag->reasmCnt; i++) {
695  while (!MSG_BOX_EMPTY(&pUnm->pMsgboxReturn[i])) {
696  NtNetBuf_t hNetBuf1;
697  hNetBuf1 = MSG_BOX_GET(&pUnm->pMsgboxReturn[i]);
698  // printf("Releasing unmatched fragment %i (%llx) by thread index %i\n", i, hNetBuf1, pUnm->streamId - pUnm->pIpDefrag->unmStart);
699  NT_NetRxRelease(hNetRx, hNetBuf1);
700  }
701  }
702 
703  /* Check for entries in the wait-list to be delivered to re-assembling thread who has signaled interest through (DOI hash table) */
704  if (pUnm->pWait) {
705  wait_list_t *pNxt, *pPrev = NULL, *pWaitFrag = pUnm->pWait;
706  if (checkTimeout) {
707  sysTs = GetSystemTimeNoArg()*100; // make it 10ns units
708  }
709  /* Loop all elements in list */
710  while (pWaitFrag) {
711  /* Try to send to an interested re-assembly tread */
712  deleteIt = SendToReasm(pUnm, &pWaitFrag->hNetBuf);
713  if (!deleteIt && checkTimeout) {
714  /* check for timed out condition */
715  if (sysTs - NT_NET_GET_PKT_TIMESTAMP(pWaitFrag->hNetBuf) > pUnm->pIpDefrag->fragTimeout) {
716  deleteIt=1;
717  }
718  }
719  /* Remove from list if send or timed out */
720  if (deleteIt) {
721  if (pWaitFrag->hNetBuf) {
722  NT_NetRxRelease(hNetRx, pWaitFrag->hNetBuf);
723  pWaitFrag->hNetBuf = NULL;
724  pUnm->fragsDeleted++;
725  }
726  pNxt = pWaitFrag->pNext;
727  /* Move base if needed */
728  if (pUnm->pWait == pWaitFrag) {
729  pUnm->pWait = pNxt;
730  } else
731  if (pPrev) {
732  pPrev->pNext = pWaitFrag->pNext;
733  }
734  /* Put onto free list */
735  pWaitFrag->pNext = pUnm->pFree;
736  pUnm->pFree = pWaitFrag;
737  /* Step to next element */
738  pWaitFrag = pNxt;
739  } else {
740  /* Jump over element */
741  pPrev = pWaitFrag;
742  pWaitFrag = pWaitFrag->pNext;
743  }
744  }
745  }
746 
747  if (hNetBuf == NULL) {
748  /* Read new fragments from stream */
749  if ((status = NT_NetRxGet(hNetRx, &hNetBuf, 1)) != NT_SUCCESS) {
750  if ((status == NT_STATUS_TIMEOUT) || (status == NT_STATUS_TRYAGAIN)) {
751  if (pUnm->pIpDefrag->fragTimeout) checkTimeout = 1;
752  continue;
753  }
754  // Get the status code as text
755  NT_ExplainError(status, errorBuffer, sizeof(errorBuffer));
756  fprintf(stderr, "[%i] NT_NetRxGet() failed: %s\n", pUnm->streamId, errorBuffer);
757  #if defined(WIN32) || defined (WIN64)
758  { _endthreadex(0); return 0; }
759  #else
760  return (void *)NULL;
761  #endif
762  }
763  }
764 
765  assert(hNetBuf != NULL);
766 
767  /* New packet received */
768  invalidPkt = 1;
769 
770  if (NT_NET_GET_PKT_L2_FRAME_TYPE(hNetBuf) == NT_L2_FRAME_TYPE_ETHER_II &&
771  NT_NET_GET_PKT_L3_FRAME_TYPE(hNetBuf) == NT_L3_FRAME_TYPE_IPv4 &&
772  NT_NET_GET_PKT_L3_FRAGMENTED(hNetBuf)) {
773 
774  /* Packet is fragmented and valid for our use */
775  if (NT_NET_GET_PKT_DESCRIPTOR_FORMAT(hNetBuf) == 9) {
776  if (NT_NET_GET_PKT_L3_FIRST_FRAG(hNetBuf)) {
777  fprintf(stderr, "[%i] ERROR un-matched streams may never receive first fragment\n", pUnm->streamId);
778  exit(0);
779  }
780 
781  invalidPkt = 0;
782  if (SendToReasm(pUnm, &hNetBuf) == 0) {
783  /* No match found. Put entry into wait list */
784  if (pUnm->pFree == NULL) {
785  /* We wait for available slots */
786  usleep(1);
787  if (pUnm->pIpDefrag->fragTimeout) checkTimeout = 1;
788  } else {
789  wait_list_t *pNxt;
790  pNxt = pUnm->pFree->pNext;
791  pUnm->pFree->pNext = pUnm->pWait;
792  pUnm->pWait = pUnm->pFree;
793  pUnm->pFree = pNxt;
794  pUnm->pWait->hNetBuf = hNetBuf;
795  /* Buffer deliverd to wait-list */
796  hNetBuf = NULL;
797  }
798  }
799 
800  if (hNetBuf == NULL) {
801  pUnm->fragRecv++;
802  }
803  }
804  }
805 
806  if (invalidPkt && !NT_NET_GET_PKT_L3_FRAGMENTED(hNetBuf)) {
807  fprintf(stderr, "[%i] ERROR, invalid non-ipv4-fragment received on one un-matched fragment stream\n", pUnm->streamId);
808  exit(0);
809  }
810  }
811 
812 /* Exiting section */
813  pWait = pUnm->pWait;
814  while (pWait) {
815  // Release pending fragment packets
816  NT_NetRxRelease(hNetRx, pWait->hNetBuf);
817  pWait = pWait->pNext;
818  }
819 
820  /* All re-assembling threads has to close before we may close the un-matched threads.
821  We need these threads to continue running while freeing hNetBuf's sent from the un-matched streams */
822  while (pUnm->pIpDefrag->allReasmClosed == 0) {
823  usleep(1000);
824  }
825 
826  /* Empty all Return Msg Boxes */
827  for (i = 0; i < pUnm->pIpDefrag->reasmCnt; i++) {
828  while (!MSG_BOX_EMPTY(&pUnm->pMsgboxReturn[i])) {
829  NtNetBuf_t buf = MSG_BOX_GET(&pUnm->pMsgboxReturn[i]);
830  NT_NetRxRelease(hNetRx, buf);
831  }
832  }
833 
834  /* Close NT stream */
835  if ((status = NT_NetRxClose(hNetRx)) != NT_SUCCESS) {
836  NT_ExplainError(status, errorBuffer, sizeof(errorBuffer));
837  fprintf(stderr, "[%i] Un-matched: NT_NetRxGet() failed: %s\n", pUnm->streamId, errorBuffer);
838  }
839 
840 #if defined(WIN32) || defined (WIN64)
841  { _endthreadex(0); return 0; }
842 #else
843  return (void *)NULL;
844 #endif
845 }
846 
847 
848 
849 static void _FreeUnmFragment(reasmThread_t *pReasm, unsigned unmIndex, NtNetBuf_t hNetBuf)
850 {
851  msg_box_t *pMsgbox;
852  int waitCnt;
853  pMsgbox = &pReasm->pIpDefrag->pReasmReturnMsgboxes[unmIndex * pReasm->pIpDefrag->reasmCnt + (pReasm->streamId - pReasm->pIpDefrag->reasmStart)];
854 
855  waitCnt=0;
856  while (waitCnt < 100) {
857  if (MSG_BOX_FULL(pMsgbox)) {
858  usleep(1000);
859  waitCnt++;
860  } else {
861  MSG_BOX_PUT(pMsgbox, hNetBuf);
862  break;
863  }
864  };
865 
866  if (waitCnt == 100) {
867  fprintf(stderr, "[%i] Return Msg Box full - failed to put element.\n", pReasm->streamId);
868  }
869 }
870 
871 
872 
873 /*****************************************************************************
874  Free a complete re-assembled datagram
875 ******************************************************************************/
876 static void _ReleaseDgramTblEntry(reasmThread_t *pReasm, tbl_entry_t *tbl_entry, NtNetStreamRx_t hNetRx)
877 {
878  unsigned i;
879  DEL_ENTRY_FROM_TBL(pReasm->tbl, tbl_entry_t, REASSEMBLY_HASH_TBL_SIZE, tbl_entry);
880 
881  for (i = 0; i < tbl_entry->fragCnt; i++) {
882  if (tbl_entry->aFrag[i].fromUnm) {
883  /* Find Return Msg box to return NetBuf to */
884  _FreeUnmFragment(pReasm, tbl_entry->aFrag[i].unmIndex, tbl_entry->aFrag[i].hNetBuf);
885  } else {
886  NT_NetRxRelease(hNetRx, tbl_entry->aFrag[i].hNetBuf);
887  }
888  }
889 
890  /* zero out interest in DOI tbl */
891  if (tbl_entry->pDOI_Src_id) {
892  *tbl_entry->pDOI_Src_id = 0;
893  }
894  if (tbl_entry->pDOI_Dst_pr) {
895  *tbl_entry->pDOI_Dst_pr = 0;
896  }
897 
898  tbl_entry->id1 = 0;
899  tbl_entry->id2 = 0;
900  RELEASE_TBL_ENTRY(pReasm->tblFree, tbl_entry);
901 }
902 
903 /*****************************************************************************
904  Check if the datagram entry specified in tbl_entry in hash table is complete
905 ******************************************************************************/
906 static int _CheckDatagramComplete(reasmThread_t *pReasm, tbl_entry_t *tbl_entry, NtNetStreamRx_t hNetRx)
907 {
908  int size = 0, i;
909  int fstFrag = -1;
910  int lstFrag = -1;
911 
912  /* Check all fragments in this datagram entry */
913  for (i = 0; i < tbl_entry->fragCnt; i++) {
914  fstFrag = i;
915  if (tbl_entry->aFrag[i].lastFrag) {
916  lstFrag = i;
917  }
918  size += tbl_entry->aFrag[i].size;
919  }
920 
921  /* if all received, process and release it */
922  if (fstFrag >= 0 && lstFrag >= 0) {
923  /* Got both first and last frag */
924  if (tbl_entry->aFrag[lstFrag].offset + tbl_entry->aFrag[lstFrag].size == size) {
925  pReasm->datagramCompleted++;
926 
927  /* A complete datagram received and collected. Should do a checksum calculation on it to validate correctness */
928 
929  _ReleaseDgramTblEntry(pReasm, tbl_entry, hNetRx);
930  return 1;
931  }
932  }
933  return 0;
934 }
935 
936 /*****************************************************************************
937  The main re-assembling thread routine. This function is called for each stream
938  specified in the load-balance configuration. It collects packets and fragments.
939  On fragment receival, it collects them into one datagram collection for
940  re-assembling. The hNetBufs are released. The thread may receive the fragments
941  from the stream it reads from, or it receives it from a msg box from one of
942  the un-matched threads. On receival of a first-fragment, the interest of that
943  datagram is raised in the DOI table associated with the un-matched thread
944  these fragments will be sent on. A first-fragment will always be sent on the
945  ordinary data streams, never on one of the un-matched streams specified.
946 ******************************************************************************/
947 #if defined(WIN32) || defined (WIN64)
948  static unsigned __stdcall _ReassemblyThread(void *arg)
949 #else
950  static void *_ReassemblyThread(void *arg)
951 #endif
952 {
953  reasmThread_t *pReasm = (reasmThread_t *)arg;
954  char tmpBuffer[20]; // Buffer used to create NTPL expressions
955  char errorBuffer[NT_ERRBUF_SIZE]; // Error buffer
956  NtNetStreamRx_t hNetRx; // Handle to the RX stream
957  NtNetBuf_t hNetBuf=NULL; // Net buffer container
958  unsigned i,ii; // Loop counters
959  int status; // NTAPI call return value
960  uint8_t fromUnm; // Flag to indicate that the hNetBuf is sent from an un-matched thread
961 
962  uint8_t unmIndex;
963 
964  int timeoutChkCnt; // Counter to control packet timeout checking
965  uint64_t sysTs; // Variable to hold system time stamp
966  DOI_dgramId_u val, val1; // DOI datagram ID variables
967  tbl_entry_t *pTbl; // Hash table traverse helper variable
968 
969  snprintf(tmpBuffer, sizeof(tmpBuffer), "IPFReasm_%i", pReasm->streamId);
970 
971  if ((status = NT_NetRxOpen(&hNetRx, tmpBuffer, NT_NET_INTERFACE_PACKET, pReasm->streamId, -1)) != NT_SUCCESS) {
972  // Get the status code as text
973  NT_ExplainError(status, errorBuffer, sizeof(errorBuffer));
974  fprintf(stderr, "[%i] NT_NetRxOpen() failed: %s\n", pReasm->streamId, errorBuffer);
975 #if defined(WIN32) || defined (WIN64)
976  { _endthreadex(0); return 0; }
977 #else
978  return (void *)NULL;
979 #endif
980  }
981 
982  timeoutChkCnt = 0;
983  while (appRunning == 1) {
984 
985  hNetBuf=NULL;
986  /* Check all Msg boxes for fragment packets from un-matched threads */
987  for (i = 0; i < pReasm->pIpDefrag->unmCnt; i++) {
988  if (!MSG_BOX_EMPTY(&pReasm->pMsgbox[i])) {
989  hNetBuf = MSG_BOX_GET(&pReasm->pMsgbox[i]);
990  fromUnm=1;
991  unmIndex = (uint8_t)i;
992  break;
993  }
994  }
995 
996  if (hNetBuf == NULL) {
997  if ((status = NT_NetRxGet(hNetRx, &hNetBuf, 1)) != NT_SUCCESS) {
998  if ((status == NT_STATUS_TIMEOUT) || (status == NT_STATUS_TRYAGAIN)) {
999  if (pReasm->pIpDefrag->fragTimeout > 0) {
1000  if (++timeoutChkCnt >= 100) {
1001  int deleteIt = 0;
1002  /* Run through list of fragments to delete timed out fragments */
1003  sysTs = GetSystemTimeNoArg()*100; // make it 1/10ns units
1004 
1005  for (i = 0; i < REASSEMBLY_HASH_TBL_SIZE; i++) {
1006  if (pReasm->tbl[i]) {
1007  tbl_entry_t *pNxt, *tbl_entry = pReasm->tbl[i];
1008  while (tbl_entry) {
1009  deleteIt = 0;
1010  for (ii = 0; ii < tbl_entry->fragCnt; ii++) {
1011  if (sysTs - NT_NET_GET_PKT_TIMESTAMP(tbl_entry->aFrag[ii].hNetBuf) > pReasm->pIpDefrag->fragTimeout) {
1012  deleteIt=1;
1013  break;
1014  }
1015  }
1016  if (deleteIt) {
1017  pNxt = tbl_entry->pNext;
1018  _ReleaseDgramTblEntry(pReasm, tbl_entry, hNetRx);
1019  pReasm->fragmentsTimedout++;
1020  tbl_entry = pNxt;
1021  } else {
1022  tbl_entry = tbl_entry->pNext;
1023  }
1024  }
1025  }
1026  }
1027  timeoutChkCnt = 0;
1028  }
1029  }
1030  continue;
1031  }
1032 
1033  // Get the status code as text
1034  NT_ExplainError(status, errorBuffer, sizeof(errorBuffer));
1035  fprintf(stderr, "[%i] NT_NetRxGet() failed: %s\n", pReasm->streamId, errorBuffer);
1036  goto EXIT;
1037  }
1038  fromUnm = 0;
1039  unmIndex = 0;
1040  }
1041 
1042  timeoutChkCnt = 0;
1043  if (NT_NET_GET_PKT_L2_FRAME_TYPE(hNetBuf) == NT_L2_FRAME_TYPE_ETHER_II &&
1044  NT_NET_GET_PKT_L3_FRAME_TYPE(hNetBuf) == NT_L3_FRAME_TYPE_IPv4 &&
1045  NT_NET_GET_PKT_L3_FRAGMENTED(hNetBuf)) {
1046  uint32_t unmEntry;
1047  uint32_t src, dst;
1048  uint8_t prot;
1049  uint16_t id;
1050  unsigned offset;
1051  tbl_entry_t *tbl_entry=NULL;
1052  DOI_dgramTbl_t *unmTbl;
1053 
1054  if (NT_NET_GET_PKT_L3_FIRST_FRAG(hNetBuf)) pReasm->firstFragRcv++;
1055  IPV4_GET_DGRAM_ID(hNetBuf, src, dst, id, prot);
1056  val.src = src;
1057  val.id = id;
1058  val1.src = dst;
1059  val1.id = prot;
1060 
1061  LOOKUP_ENTRY(pReasm->tbl, tbl_entry_t, REASSEMBLY_HASH_TBL_SIZE, val.src_id, val1.src_id, tbl_entry);
1062  if (tbl_entry == NULL) {
1063  GET_NEW_TBL_ENTRY(pReasm->tblFree, tbl_entry_t, tbl_entry);
1064  }
1065 
1066  if (tbl_entry->id1 == 0) {
1067  // New datagram fragment received
1068  tbl_entry->id1 = val.src_id;
1069  tbl_entry->id2 = val1.src_id;
1070  tbl_entry->fragCnt = 0;
1071  tbl_entry->pDOI_Src_id = NULL;
1072  tbl_entry->pDOI_Dst_pr = NULL;
1073  ADD_ENTRY_TO_TBL(pReasm->tbl, tbl_entry_t, REASSEMBLY_HASH_TBL_SIZE, tbl_entry);
1074  } else {
1075  // Error check
1076  if (tbl_entry->fragCnt == MAX_FRAG_CNT) {
1077  fprintf(stderr, "[%i] ERROR - MAX frag cnt reached\n", pReasm->streamId);
1078  exit(0);
1079  }
1080  }
1081 
1082  // Add fragment to datagram in list
1083  tbl_entry->aFrag[tbl_entry->fragCnt].hNetBuf = hNetBuf;
1084  tbl_entry->aFrag[tbl_entry->fragCnt].fromUnm = fromUnm;
1085  tbl_entry->aFrag[tbl_entry->fragCnt].unmIndex = unmIndex;
1086 
1087 /*
1088 * warning You may get conversion warnings here caused by ntohs(). This is a bug in /usr/include/bits/byteswap.h
1089 */
1090  tbl_entry->aFrag[tbl_entry->fragCnt].offset = (uint16_t)IPV4_FRAGMENT_OFFSET(hNetBuf);
1091  tbl_entry->aFrag[tbl_entry->fragCnt].size = (uint16_t)IPV4_DATA_LEN(hNetBuf);
1092  tbl_entry->aFrag[tbl_entry->fragCnt].firstFrag = (uint8_t)NT_NET_GET_PKT_L3_FIRST_FRAG(hNetBuf);
1093 
1094  if (pReasm->pIpDefrag->extDescrType == 9) {
1095  tbl_entry->aFrag[tbl_entry->fragCnt].lastFrag = (uint8_t)NT_NET_GET_PKT_IPF_LAST_FRAGMENT(hNetBuf);
1096  } else {
1097  tbl_entry->aFrag[tbl_entry->fragCnt].lastFrag = IPV4_LAST_FRAG(hNetBuf);
1098  }
1099 
1100  tbl_entry->fragCnt++;
1101 
1102  /* Check datagram for completion */
1103  if (_CheckDatagramComplete(pReasm, tbl_entry, hNetRx) == 0) {
1104  if (NT_NET_GET_PKT_DESCRIPTOR_FORMAT(hNetBuf) == 9) {
1105  if (NT_NET_GET_PKT_L3_FIRST_FRAG(hNetBuf)) {
1106  int idx=-1;
1107 
1108  /* Find thread-index of the un-matched stream receiving un-matched fragments */
1109  offset = (unsigned)NT_NET_GET_PKT_IPF_UNMATCHED_STREAMID(hNetBuf) - pReasm->pIpDefrag->unmStart;
1110  /* Get a pointer to the DOI table between this re-assembling thread and the un-matching fragments thread */
1111  unmTbl = &pReasm->pIpDefrag->pDOI_Tbls[(offset * pReasm->pIpDefrag->reasmCnt + pReasm->idx)* DOI_FRAG_TBL_SIZE];
1112  DOI_HASH_GET_KEY(unmEntry, src, id);
1113 
1114  // On first fragment receival - Notify the corresponding un-matched thread about interest in fragments of this datagram
1115  for (i=0;i<MAX_SRC_ID;i++) {
1116  if (unmTbl[unmEntry].aSrc_id[i] == val.src_id &&
1117  unmTbl[unmEntry].aDst_pr[i] == val1.src_id ) {
1118  pReasm->Src_IdClash++;
1119  /* Ok already there! */
1120  idx = (int)i;
1121  break;
1122  }
1123  if (idx < 0 && unmTbl[unmEntry].aSrc_id[i] == 0) {
1124  idx = (int)i;
1125  }
1126  }
1127 
1128  if (idx >= 0) {
1129  tbl_entry->pDOI_Src_id = &unmTbl[unmEntry].aSrc_id[idx];
1130  tbl_entry->pDOI_Dst_pr = &unmTbl[unmEntry].aDst_pr[idx];
1131  *tbl_entry->pDOI_Dst_pr = val1.src_id;
1132  *tbl_entry->pDOI_Src_id = val.src_id;
1133  } else {
1134  fprintf(stderr, "[%i] Too many clashes in DOI (more than %i - raise MAX_SRC_ID)\n", pReasm->streamId, MAX_SRC_ID);
1135  exit(0);
1136  }
1137  }
1138  }
1139  }
1140 
1141  } else {
1142  pReasm->nonFragments++;
1143 
1144  // Release packet as it is not fragmented.
1145  if ((status = NT_NetRxRelease(hNetRx, hNetBuf)) != NT_SUCCESS) {
1146  // Get the status code as text
1147  NT_ExplainError(status, errorBuffer, sizeof(errorBuffer));
1148  fprintf(stderr, "[%i] NT_NetRxRelease() failed: %s\n", pReasm->streamId , errorBuffer);
1149  goto EXIT;
1150  }
1151  }
1152  }
1153 
1154 EXIT:
1155  for (i = 0; i < REASSEMBLY_HASH_TBL_SIZE; i++) {
1156  pTbl = pReasm->tbl[i];
1157  while (pTbl) {
1158  // Release all hash table fragment packets
1159  for (ii = 0; ii < pTbl->fragCnt; ii++) {
1160  if (pTbl->aFrag[ii].fromUnm) {
1161  _FreeUnmFragment(pReasm, pTbl->aFrag[ii].unmIndex, pTbl->aFrag[ii].hNetBuf);
1162  } else {
1163  NT_NetRxRelease(hNetRx, pTbl->aFrag[ii].hNetBuf);
1164  }
1165  }
1166  pTbl = pTbl->pNext;
1167  }
1168  }
1169 
1170  /* Empty Msg boxes */
1171  for (i = 0; i < pReasm->pIpDefrag->unmCnt; i++) {
1172  while (!MSG_BOX_EMPTY(&pReasm->pMsgbox[i])) {
1173  hNetBuf = MSG_BOX_GET(&pReasm->pMsgbox[i]);
1174  _FreeUnmFragment(pReasm, i, hNetBuf);
1175  pReasm->msgboxPackets++;
1176  }
1177  }
1178 
1179  // Close the stream
1180  if ((status = NT_NetRxClose(hNetRx)) != NT_SUCCESS) {
1181  NT_ExplainError(status, errorBuffer, sizeof(errorBuffer));
1182  fprintf(stderr, "[%i] Re-asm thread: NT_NetRxGet() failed: %s\n", pReasm->streamId, errorBuffer);
1183  }
1184 #if defined(WIN32) || defined (WIN64)
1185  { _endthreadex(0); return 0; }
1186 #else
1187  return (void *)NULL;
1188 #endif
1189 }
1190 
1191 /*****************************************************************************
1192  Ctrl-C signal handler routine.
1193 ******************************************************************************/
1194 #if defined(WIN32) || defined (WIN64)
1195 static BOOL WINAPI StopApplication(int sig)
1196 {
1197  (void) sig;
1198  appRunning = 0;
1199  return TRUE;
1200 }
1201 #else
1202 static void StopApplication(int sig)
1203 {
1204  if (sig == SIGINT)
1205  appRunning = 0;
1206 }
1207 #endif
1208 
1209 /* Default paramters */
1210 #define INPUT_ADAPTER 0
1211 #define FRAGMENT_TIMEOUT 2000 // mSec
1212 #define NUM_STREAMS 4
1213 #define NUM_UNM_STREAMS 4
1214 #define TABLE_TIMEOUT 125
1215 #define TABLE_PERSIST_TIMEOUT 0
1216 
1217 /*****************************************************************************
1218  Main routine. It initializes, configures FPGA, starts needed threads
1219  (un-matched and re-assembling threads), and loops while printing
1220  out statistics. On Ctrl-C it stops looping and begins the cleanup process.
1221 ******************************************************************************/
1222 int main(int argc, const char *argv[])
1223 {
1224  char tmpBuffer[100]; // Buffer to build filter string
1225  char errorBuffer[NT_ERRBUF_SIZE]; // Error buffer
1226  int status; // Status variable
1227  NtNtplInfo_t ntplInfo; // Return data structure from the NT_NTPL() call.
1228  NtInfoStream_t hInfoStream; // Info stream handle
1229  NtInfo_t hInfo; // Info handle
1230  unsigned i, ii; // Counter variables
1231  int inp_1, inp_2; // Input port numbers for selected adapter
1232  NtConfigStream_t hCfgStream; // Handle to a config stream
1233  int hashTblFreed; // Counter for re-assemble hash table entries remained on exit
1234  int waitTblFreed; // Counter for un-matched wait table entries remained on exit
1235  int msgboxElmFreed; // Counter for hNetBuf elements remained in msg boxes
1236  int tablePersistTimeout; // Control parameter for NTPL IPFMode TablePersist setting
1237  int tableTimeout; // Control parameter for NTPL IPFMode timeout setting
1238  struct argparse argparse;
1239 #define NUM_NTPL 3
1240  const char *ntplExpr[NUM_NTPL] =
1241  {"Assign[streamid=(%i..%i)]=port==(%i..%i)",
1242  "HashMode=Hash%iTuple",
1243  "IPFMode[StreamId=(%i..%i);timeout=%i;TablePersist=%s]=port==(%i..%i)"};
1244 
1245  // Set up ctrl+c handler
1246 #if defined(WIN32) || defined (WIN64)
1247  SetConsoleCtrlHandler((PHANDLER_ROUTINE)StopApplication, TRUE);
1248 #else
1249  struct sigaction newaction; // Ctrl+c handle
1250  memset(&newaction, 0, sizeof(newaction));
1251  newaction.sa_handler = StopApplication;
1252  if (sigaction(SIGINT, &newaction, NULL) < 0) {
1253  fprintf(stderr, "Failed to register SIGINT sigaction.\n");
1254  exit(EXIT_FAILURE);
1255  }
1256 #endif
1257 
1258  // Setup default parameter settings
1259  memset(&IpDefrag, 0 , sizeof(struct _ipDefrag));
1260  IpDefrag.adapterNo = INPUT_ADAPTER;
1261  IpDefrag.reasmCnt = NUM_STREAMS;
1262  IpDefrag.reasmStart = 0;
1263  IpDefrag.unmCnt = NUM_UNM_STREAMS;
1264  IpDefrag.unmStart = IpDefrag.reasmStart + IpDefrag.reasmCnt;
1265  IpDefrag.fragTimeout = FRAGMENT_TIMEOUT * 100000ULL;
1266  tableTimeout = TABLE_TIMEOUT * 10;
1267  tablePersistTimeout = TABLE_PERSIST_TIMEOUT;
1268 
1269  argparse_init(&argparse, arg_options, usageText, 0);
1270  argparse_parse(&argparse, argc, argv);
1271 
1272  if (opt_adapter != -1) {
1273  IpDefrag.adapterNo = opt_adapter;
1274  }
1275  if (opt_reasm != -1) {
1276  IpDefrag.reasmCnt = (unsigned)opt_reasm;
1277  }
1278  if (opt_unm != -1) {
1279  IpDefrag.unmCnt = (unsigned)opt_unm;
1280  }
1281  if (opt_frag != -1) {
1282  IpDefrag.fragTimeout = (unsigned)opt_frag;
1283  IpDefrag.fragTimeout *= 100000U;
1284  }
1285  if (opt_persist != NULL) {
1286  if (strcmp(opt_persist, "timeout") == 0) {
1287  tablePersistTimeout = 1;
1288  }
1289  }
1290  if (opt_timeout != -1) {
1291  tableTimeout = opt_timeout * 10;
1292  }
1293 
1294  // Initialize the NTAPI library and thereby check if NTAPI_VERSION can be used together with this library
1295  if ((status = NT_Init(NTAPI_VERSION)) != NT_SUCCESS) {
1296  // Get the status code as text
1297  NT_ExplainError(status, errorBuffer, sizeof(errorBuffer));
1298  fprintf(stderr, "NT_Init() failed: %s\n", errorBuffer);
1299  return -1;
1300  }
1301 
1302  /* Open the info stream */
1303  if ((status = NT_InfoOpen(&hInfoStream, "IPFExample")) != NT_SUCCESS) {
1304  // Get the status code as text
1305  NT_ExplainError(status, errorBuffer, sizeof(errorBuffer));
1306  fprintf(stderr, "NT_InfoOpen() failed: %s\n", errorBuffer);
1307  return -1;
1308  }
1309 
1310  /* Read number of adapter */
1311  hInfo.cmd=NT_INFO_CMD_READ_ADAPTER_V7;
1312  hInfo.u.adapter_v7.adapterNo = (uint8_t) IpDefrag.adapterNo;
1313 
1314  if ((status = NT_InfoRead(hInfoStream, &hInfo)) != 0) {
1315  // Get the status code as text
1316  NT_ExplainError(status, errorBuffer, sizeof(errorBuffer));
1317  fprintf(stderr, "NT_InfoRead() failed: %s\n", errorBuffer);
1318  return -1;
1319  }
1320  /* collect port range for the selected adapter */
1321  inp_1 = hInfo.u.adapter_v7.data.portOffset;
1322  inp_2 = hInfo.u.adapter_v7.data.portOffset + hInfo.u.adapter_v7.data.numPorts - 1;
1323 
1324  /* Ensure that Ext9 was configured. */
1325  if (NT_PACKET_DESCRIPTOR_TYPE_NT_EXTENDED == hInfo.u.adapter_v7.data.descriptorType && 9 == hInfo.u.adapter_v7.data.extendedDescriptor) {
1326  IpDefrag.extDescrType = hInfo.u.adapter_v7.data.extendedDescriptor;
1327  } else {
1328  fprintf(stderr, "The packet descriptor is not Ext9. Please set 'PacketDescriptor = Ext9' in ntservice.ini file for the selected adapter.\n");
1329  return -1;
1330  }
1331 
1332  /* Get timesync configuration. Must use OS mode to run timeouts with this demo */
1333  if (IpDefrag.fragTimeout) {
1334  hInfo.cmd=NT_INFO_CMD_READ_TIMESYNC_V4;
1335  hInfo.u.timeSync_v4.adapterNo = (uint8_t) IpDefrag.adapterNo;
1336  if ((status = NT_InfoRead(hInfoStream, &hInfo)) != NT_SUCCESS) {
1337  // Get the status code as text
1338  NT_ExplainError(status, errorBuffer, sizeof(errorBuffer));
1339  fprintf(stderr, "Failed to read timesync info: %s\n", errorBuffer);
1340  return -1;
1341  }
1342 
1343  if (hInfo.u.timeSync_v4.data.timeRef != NT_TIMESYNC_REFERENCE_OSTIME) {
1344  fprintf(stderr, "The timesync reference clock on the selected adapter is not OS. Please set TimeSyncReferencePriority=OsTime in ntservice.ini file for the selected adapter\n");
1345  return -1;
1346  }
1347  }
1348 
1349  /* Close info stream */
1350  if ((status = NT_InfoClose(hInfoStream)) != NT_SUCCESS) {
1351  // Get the status code as text
1352  NT_ExplainError(status, errorBuffer, sizeof(errorBuffer));
1353  fprintf(stderr, "NT_InfoClose() failed: %s\n", errorBuffer);
1354  return -1;
1355  }
1356 
1357  // Open a config stream to assign a filter to a stream ID.
1358  if ((status = NT_ConfigOpen(&hCfgStream, "IPFExample")) != NT_SUCCESS) {
1359  // Get the status code as text
1360  NT_ExplainError(status, errorBuffer, sizeof(errorBuffer));
1361  fprintf(stderr, "NT_ConfigOpen() failed: %s\n", errorBuffer);
1362  return -1;
1363  }
1364 
1365  /* Build final NTPL command strings */
1366  for (i = 0; i < NUM_NTPL; i++) {
1367  switch (i) {
1368  case 0: snprintf(tmpBuffer, sizeof(tmpBuffer), ntplExpr[i], IpDefrag.reasmStart, IpDefrag.reasmStart + IpDefrag.reasmCnt - 1, inp_1, inp_2); break;
1369  case 1: snprintf(tmpBuffer, sizeof(tmpBuffer), ntplExpr[i], (IpDefrag.extDescrType == 9)?5:2); break;
1370  case 2:
1371  if (IpDefrag.extDescrType == 9) {
1372  if (tablePersistTimeout) {
1373  snprintf(tmpBuffer, sizeof(tmpBuffer), ntplExpr[i], IpDefrag.unmStart, IpDefrag.unmStart + IpDefrag.unmCnt - 1, tableTimeout, "TimeoutOnly", inp_1, inp_2);
1374  } else {
1375  snprintf(tmpBuffer, sizeof(tmpBuffer), ntplExpr[i], IpDefrag.unmStart, IpDefrag.unmStart + IpDefrag.unmCnt - 1, tableTimeout, "LastFragment", inp_1, inp_2);
1376  }
1377  } else {
1378  IpDefrag.unmCnt = 0;
1379  continue;
1380  }
1381  break;
1382  default:
1383  strncpy(tmpBuffer, ntplExpr[i], sizeof(tmpBuffer) - 1);
1384  tmpBuffer[sizeof(tmpBuffer) - 1] = '\0';
1385  break;
1386  }
1387 
1388  // Assign NTPL expressions needed
1389  if ((status = NT_NTPL(hCfgStream, tmpBuffer, &ntplInfo, NT_NTPL_PARSER_VALIDATE_NORMAL)) != NT_SUCCESS) {
1390  fprintf(stderr, "ERROR --> %s\n", tmpBuffer);
1391  // Get the status code as text
1392  NT_ExplainError(status, errorBuffer, sizeof(errorBuffer));
1393  fprintf(stderr, "NT_NTPL() failed: %s\n", errorBuffer);
1394  fprintf(stderr, ">>> NTPL errorcode: %X\n", ntplInfo.u.errorData.errCode);
1395  fprintf(stderr, ">>> %s\n", ntplInfo.u.errorData.errBuffer[0]);
1396  fprintf(stderr, ">>> %s\n", ntplInfo.u.errorData.errBuffer[1]);
1397  fprintf(stderr, ">>> %s\n", ntplInfo.u.errorData.errBuffer[2]);
1398  return -1;
1399  }
1400  }
1401 
1402  appRunning = 1;
1403  IpDefrag.allReasmClosed = 0;
1404 
1405  /* Create IPv4 Reassembly threads */
1406  IpDefrag.pReasmThreads = calloc(1, sizeof(reasmThread_t) * IpDefrag.reasmCnt);
1407  if (IpDefrag.pReasmThreads == NULL) {
1408  fprintf(stderr, "Memory allocation failed\n");
1409  return -1;
1410  }
1411 
1412  /* Create communication message boxes. One msg box for each Rasm thread from each Unm thread */
1413  /* Create twice - one for NetBuf send from unmatched thread to reassembling thread and one for */
1414  /* returning the NetBuf again to the unmatched thread */
1415  IpDefrag.pReasmMsgboxes =
1416  calloc(1, sizeof(msg_box_t) * IpDefrag.reasmCnt * 2 *
1417  /* IpDefrag.unmCnt == 0 when IpDefrag.extDescrType != 8 */
1418  (IpDefrag.unmCnt == 0 ? 1 : IpDefrag.unmCnt));
1419 
1420  if (IpDefrag.pReasmMsgboxes == NULL) {
1421  fprintf(stderr, "Memory allocation failed\n");
1422  return -1;
1423  }
1424  IpDefrag.pReasmReturnMsgboxes = (msg_box_t *)((char *)IpDefrag.pReasmMsgboxes + sizeof(msg_box_t) * IpDefrag.reasmCnt * IpDefrag.unmCnt);
1425 
1426  /* initialize and start all IP fragment re-assembling threads */
1427  for (i = 0; i < IpDefrag.reasmCnt; i++) {
1428  tbl_entry_t *tbl;
1429 
1430  IpDefrag.pReasmThreads[i].pIpDefrag = &IpDefrag;
1431  IpDefrag.pReasmThreads[i].streamId = IpDefrag.reasmStart + i;
1432  IpDefrag.pReasmThreads[i].idx = i;
1433  IpDefrag.pReasmThreads[i].pMsgbox = &IpDefrag.pReasmMsgboxes[i * IpDefrag.unmCnt];
1434 
1435  /* Create initial hash table elements and fill in free list. May grow dynamically later */
1436  IpDefrag.pReasmThreads[i].tblFree = calloc(1, sizeof(tbl_entry_t));
1437  tbl = IpDefrag.pReasmThreads[i].tblFree;
1438  if (tbl) {
1439  for (ii = 1; ii < INITIAL_HASH_TBL_ENTRY_CNT; ii++) {
1440  tbl->pNext = calloc(1, sizeof(tbl_entry_t));
1441  tbl=tbl->pNext;
1442  if (tbl == NULL) break;
1443  }
1444  }
1445 
1446  if ((status = pthread_create(&IpDefrag.pReasmThreads[i].thread, NULL, _ReassemblyThread, (void*)&IpDefrag.pReasmThreads[i])) != 0) {
1447  fprintf(stderr, "Unable to create Unmatched stream thread");
1448  return status;
1449  }
1450  }
1451 
1452  if (IpDefrag.unmCnt) {
1453  /* Create all DOI-tables for all un-matched threads */
1454  IpDefrag.pDOI_Tbls = calloc(1, sizeof(DOI_dgramTbl_t) * DOI_FRAG_TBL_SIZE * IpDefrag.unmCnt * IpDefrag.reasmCnt);
1455 
1456  /* Create Unmatched fragment threads */
1457  IpDefrag.pUnmThreads = calloc(1, sizeof(unmThread_t) * IpDefrag.unmCnt);
1458  if (IpDefrag.pUnmThreads == NULL) {
1459  fprintf(stderr, "Memory allocation failed\n");
1460  return -1;
1461  }
1462 
1463  for (i = 0; i < IpDefrag.unmCnt; i++) {
1464  IpDefrag.pUnmThreads[i].pIpDefrag = &IpDefrag;
1465  IpDefrag.pUnmThreads[i].streamId = IpDefrag.unmStart + i;
1466  IpDefrag.pUnmThreads[i].pDOI = &IpDefrag.pDOI_Tbls[i * DOI_FRAG_TBL_SIZE * IpDefrag.reasmCnt];
1467  IpDefrag.pUnmThreads[i].pMsgboxReturn = &IpDefrag.pReasmReturnMsgboxes[i * IpDefrag.reasmCnt];
1468 
1469  if ((status = pthread_create(&IpDefrag.pUnmThreads[i].thread, NULL, _UnMatchedThread, (void*)&IpDefrag.pUnmThreads[i])) != 0) {
1470  fprintf(stderr, "Unable to create Unmatched stream thread");
1471  return status;
1472  }
1473  }
1474  }
1475 
1476  /* main statistics loop */
1477  while (appRunning == 1) {
1478  int sum1,sum2,sum3,sum4,sum5;
1479  printf("Parameters:\n");
1480  printf("Adapter number to run on %i\n", IpDefrag.adapterNo);
1481  printf("Number of streams/threads %i\n", IpDefrag.reasmCnt);
1482  printf("Number of un-matched streams/threads %i\n", IpDefrag.unmCnt);
1483  printf("Fragment timeout (ms) %i\n", (int)(IpDefrag.fragTimeout/100000));
1484  printf("IPFMode table timeout (ms) %i\n", tableTimeout/10);
1485  printf("IPFMode TablePersist ");
1486  if (tablePersistTimeout) {
1487  printf("TimeoutOnly\n");
1488  } else {
1489  printf("LastFragment\n");
1490  }
1491  printf("\nRunning with extended descriptor %i\n", IpDefrag.extDescrType);
1492  printf("-------------------------------------------------------------------\n");
1493 
1494  printf("Re-assembling streams:\n");
1495  printf(" # Completed First frag Deleted partial Non-fragmented Id\n");
1496  printf(" (timed out) packets clashes\n");
1497 
1498  sum1 = sum2 = sum3 = sum4 = sum5 = 0;
1499  for (i = 0; i < IpDefrag.reasmCnt; i++) {
1500  printf("%2d:%8d, %8d, %10d, %14d, %12d\n", IpDefrag.pReasmThreads[i].streamId,
1501  IpDefrag.pReasmThreads[i].datagramCompleted, IpDefrag.pReasmThreads[i].firstFragRcv,
1502  IpDefrag.pReasmThreads[i].fragmentsTimedout, IpDefrag.pReasmThreads[i].nonFragments,
1503  IpDefrag.pReasmThreads[i].Src_IdClash);
1504  sum1 += IpDefrag.pReasmThreads[i].datagramCompleted;
1505  sum2 += IpDefrag.pReasmThreads[i].firstFragRcv;
1506  sum3 += IpDefrag.pReasmThreads[i].fragmentsTimedout;
1507  sum4 += IpDefrag.pReasmThreads[i].nonFragments;
1508  sum5 += IpDefrag.pReasmThreads[i].Src_IdClash;
1509  }
1510  printf("sum:%7d, %8d, %10d, %14d, %12d\n", sum1,sum2,sum3,sum4,sum5);
1511 
1512  if (IpDefrag.unmCnt) {
1513  printf("Un-matched fragment streams:\n");
1514  printf(" # Fragments Deleted frags\n");
1515  printf(" received (timed out)\n");
1516  for (i = 0; i< IpDefrag.unmCnt; i++) {
1517  printf("%2d:%8d,%12d\n", IpDefrag.pUnmThreads[i].streamId, IpDefrag.pUnmThreads[i].fragRecv, IpDefrag.pUnmThreads[i].fragsDeleted);
1518  }
1519  }
1520  printf("\n-------------------------------------------------------------------\n");
1521  fflush(stdout);
1522  // Sleep 2 sec
1523  sleep(2);
1524  }
1525 
1526  /* Cleanup section */
1527  hashTblFreed = 0;
1528  waitTblFreed = 0;
1529  msgboxElmFreed = 0;
1530  for (i = 0; i < IpDefrag.reasmCnt; i++) {
1531  tbl_entry_t *tbl;
1532  pthread_join(IpDefrag.pReasmThreads[i].thread, NULL);
1533  /* Free all free table entries */
1534  while (IpDefrag.pReasmThreads[i].tblFree) {
1535  tbl = IpDefrag.pReasmThreads[i].tblFree->pNext;
1536  free(IpDefrag.pReasmThreads[i].tblFree);
1537  IpDefrag.pReasmThreads[i].tblFree = tbl;
1538  }
1539  /* Free all hash table entries containing fragments */
1540  for (ii = 0; ii < REASSEMBLY_HASH_TBL_SIZE; ii++) {
1541  while (IpDefrag.pReasmThreads[i].tbl[ii]) {
1542  tbl = IpDefrag.pReasmThreads[i].tbl[ii]->pNext;
1543  hashTblFreed++; // already released on thread termination
1544  free(IpDefrag.pReasmThreads[i].tbl[ii]);
1545  IpDefrag.pReasmThreads[i].tbl[ii] = tbl;
1546  }
1547  }
1548  msgboxElmFreed += IpDefrag.pReasmThreads[i].msgboxPackets;
1549  }
1550  free(IpDefrag.pReasmThreads);
1551 
1552  IpDefrag.allReasmClosed = 1;
1553 
1554  for (i = 0; i < IpDefrag.unmCnt; i++) {
1555  pthread_join(IpDefrag.pUnmThreads[i].thread, NULL);
1556  }
1557 
1558  for (i = 0; i < IpDefrag.unmCnt; i++) {
1559  wait_list_t *pWait = IpDefrag.pUnmThreads[i].pWait;
1560  while (pWait) {
1561  waitTblFreed++; // already released on thread termination
1562  pWait=pWait->pNext;
1563  }
1564  }
1565 
1566  free(IpDefrag.pUnmThreads);
1567  free(IpDefrag.pReasmMsgboxes);
1568  free(IpDefrag.pDOI_Tbls);
1569 
1570  if (msgboxElmFreed || hashTblFreed || waitTblFreed) {
1571  printf("\n----------------------------------------------------------");
1572  if (hashTblFreed) {
1573  printf("\n%i entries remained in hash table on exit", hashTblFreed);
1574  }
1575  if (waitTblFreed) {
1576  printf("\n%i entries remained in un-matched fragments wait-table on exit", waitTblFreed);
1577  }
1578  if (msgboxElmFreed) {
1579  printf("\n%i entries remained in msg boxes on exit", msgboxElmFreed);
1580  }
1581  printf("\n----------------------------------------------------------\n");
1582  }
1583 
1584  // Delete the filter
1585  snprintf(tmpBuffer, sizeof(tmpBuffer), "delete=%d", ntplInfo.ntplId);
1586  if ((status = NT_NTPL(hCfgStream, tmpBuffer, &ntplInfo, NT_NTPL_PARSER_VALIDATE_NORMAL)) != NT_SUCCESS) {
1587  // Get the status code as text
1588  NT_ExplainError(status, errorBuffer, sizeof(errorBuffer));
1589  fprintf(stderr, "NT_NTPL() failed: %s\n", errorBuffer);
1590  fprintf(stderr, ">>> NTPL errorcode: %X\n", ntplInfo.u.errorData.errCode);
1591  fprintf(stderr, ">>> %s\n", ntplInfo.u.errorData.errBuffer[0]);
1592  fprintf(stderr, ">>> %s\n", ntplInfo.u.errorData.errBuffer[1]);
1593  fprintf(stderr, ">>> %s\n", ntplInfo.u.errorData.errBuffer[2]);
1594  return -1;
1595  }
1596 
1597  // Close the config stream
1598  if ((status = NT_ConfigClose(hCfgStream)) != NT_SUCCESS) {
1599  // Get the status code as text
1600  NT_ExplainError(status, errorBuffer, sizeof(errorBuffer));
1601  fprintf(stderr, "NT_ConfigClose() failed: %s\n", errorBuffer);
1602  return -1;
1603  }
1604  return 0;
1605 }