Add Express Mesh as a network option.

-- Some commits include fixes in other networks

codes/model-net-lp.h

@@ -29,6 +29,7 @@ extern "C" {
 #include "net/simplenet-upd.h"
 #include "net/simplep2p.h"
 #include "net/torus.h"
+#include "net/express-mesh.h"
 
 extern int model_net_base_magic;
 
@@ -132,6 +133,7 @@ typedef struct model_net_wrap_msg {
         sn_message              m_snet;  // simplenet
         sp_message              m_sp2p;  // simplep2p
         nodes_message           m_torus; // torus
+        em_message              m_em; // express-mesh
         // add new ones here
     } msg;
 } model_net_wrap_msg;

codes/model-net.h

@@ -72,6 +72,8 @@ typedef struct mn_stats mn_stats;
     X(DRAGONFLY_CUSTOM, "modelnet_dragonfly_custom", "dragonfly_custom", &dragonfly_custom_method)\
     X(DRAGONFLY_CUSTOM_ROUTER, "modelnet_dragonfly_custom_router", "dragonfly_custom_router", &dragonfly_custom_router_method)\
     X(LOGGP,     "modelnet_loggp",     "loggp",     &loggp_method)\
+    X(EXPRESS_MESH, "modelnet_express_mesh", "express_mesh", &express_mesh_method)\
+    X(EXPRESS_MESH_ROUTER, "modelnet_express_mesh_router", "express_mesh_router", &express_mesh_router_method)\
     X(MAX_NETS,  NULL,                 NULL,        NULL)
 
 #define X(a,b,c,d) a,

codes/net/common-net.h

+#ifndef COMMON_NET_H
+#define COMMON_NET_H
+#include "codes/model-net-lp.h"
+#include "codes/quickhash.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+
+struct mn_hash_key
+{
+  uint64_t message_id;
+  tw_lpid sender_id;
+};
+
+struct mn_qhash_entry
+{
+  struct mn_hash_key key;
+  char * remote_event_data;
+  int num_chunks;
+  int remote_event_size;
+  struct qhash_head hash_link;
+};
+
+extern int mn_rank_hash_compare(void *key, struct qhash_head *link);
+
+extern int mn_hash_func(void *k, int table_size);
+
+extern void free_tmp(void * ptr);
+
+typedef struct message_list message_list;
+
+struct message_list {
+  union {
+    terminal_message dfly_msg;
+    em_message em_msg;
+  };
+  char* event_data;
+  message_list *next;
+  message_list *prev;
+  int port, index;
+  message_list *altq_next, *altq_prev;
+  int in_alt_q, altq_port;
+};
+
+extern void append_to_message_list(  
+    message_list ** thisq,
+    message_list ** thistail,
+    int index, 
+    message_list *msg);
+
+extern void prepend_to_message_list(  
+    message_list ** thisq,
+    message_list ** thistail,
+    int index, 
+    message_list *msg);
+
+extern message_list* return_head(
+    message_list ** thisq,
+    message_list ** thistail,
+    int index);
+
+extern message_list* return_tail(
+    message_list ** thisq,
+    message_list ** thistail,
+    int index);
+
+extern void delete_from_message_list(
+    message_list *** allq, 
+    message_list *** alltail,
+    message_list *msg);
+
+extern void add_to_message_list(
+    message_list *** allq, 
+    message_list *** alltail,
+    message_list *msg);
+
+extern void altq_append_to_message_list(  
+    message_list ** thisq,
+    message_list ** thistail,
+    int index, 
+    message_list *msg);
+
+extern void altq_prepend_to_message_list(  
+    message_list ** thisq,
+    message_list ** thistail,
+    int index, 
+    message_list *msg);
+
+extern message_list* altq_return_head(
+    message_list ** thisq,
+    message_list ** thistail,
+    int index);
+
+extern message_list* altq_return_tail(
+    message_list ** thisq,
+    message_list ** thistail,
+    int index);
+
+extern void altq_delete_from_message_list(
+    message_list *** allq, 
+    message_list *** alltail,
+    message_list *msg);
+
+extern void altq_add_to_message_list(
+    message_list *** allq, 
+    message_list *** alltail,
+    message_list *msg);
+
+extern void delete_message_list(void *thism);
+extern tw_stime bytes_to_ns(uint64_t bytes, double GB_p_s);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

codes/net/express-mesh.h

+/*
+ * Copyright (C) 2014 University of Chicago.
+ * See COPYRIGHT notice in top-level directory.
+ *
+ */
+
+#ifndef EXPRESS_MESH_H
+#define EXPRESS_MESH_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <ross.h>
+
+typedef struct em_message em_message;
+
+struct em_message
+{
+  /* magic number */
+  int magic;
+  /* flit travel start time*/
+  tw_stime travel_start_time;
+ /* packet ID of the flit  */
+  unsigned long long packet_ID;
+  /* event type of the flit */
+  short  type;
+  /* category: comes from codes */
+  char category[CATEGORY_NAME_MAX];
+
+  /* final destination LP ID, this comes from codes can be a server or any other LP type*/
+  tw_lpid final_dest_gid;
+  /*sending LP ID from CODES, can be a server or any other LP type */
+  tw_lpid sender_lp;
+  tw_lpid sender_mn_lp; // source modelnet id
+ /* destination terminal ID */
+  tw_lpid dest_terminal_id;
+  int dest_terminal;
+  /* source terminal ID */
+  tw_lpid src_terminal_id;
+
+  short saved_channel;
+  short my_N_hop;
+  short hops[8];
+
+  /* Intermediate LP ID from which this message is coming */
+  unsigned int intm_lp_id;
+  short saved_vc;
+  short dim_change;
+  /* last hop of the message, can be a terminal, local router or global router */
+  int last_hop;
+  /* For routing */
+  uint64_t chunk_id;
+  uint64_t packet_size;
+  uint64_t message_id;
+  uint64_t total_size;
+
+  int saved_remote_esize;
+  int remote_event_size_bytes;
+  int local_event_size_bytes;
+
+  // For buffer message
+  int vc_index;
+  int output_chan;
+  model_net_event_return event_rc;
+  int is_pull;
+  uint64_t pull_size;
+
+  /* for reverse computation */   
+  tw_stime saved_available_time;
+  tw_stime saved_avg_time;
+  tw_stime saved_rcv_time;
+  tw_stime saved_busy_time; 
+  tw_stime saved_total_time;
+  tw_stime saved_hist_start_time;
+  tw_stime saved_sample_time;
+  tw_stime msg_start_time;
+
+  int saved_hist_num;
+  int saved_occupancy;
+};
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif 
+

src/Makefile.subdir

@@ -87,6 +87,7 @@ nobase_include_HEADERS = \
 	codes/model-net-lp.h \
 	codes/model-net-sched.h \
 	codes/model-net-inspect.h \
+	codes/net/common-net.h \
 	codes/net/dragonfly.h \
 	codes/net/dragonfly-custom.h \
 	codes/net/slimfly.h \
@@ -94,6 +95,7 @@ nobase_include_HEADERS = \
 	codes/net/loggp.h \
 	codes/net/simplenet-upd.h \
 	codes/net/simplep2p.h \
+	codes/net/express-mesh.h \
 	codes/net/torus.h \
   codes/codes-mpi-replay.h \
 	codes/configfile.h
@@ -149,8 +151,10 @@ src_libcodes_la_SOURCES =  \
 	codes/rc-stack.h \
 	src/util/rc-stack.c \
 	src/networks/model-net/model-net.c \
+	src/networks/model-net/common-net.c \
 	src/networks/model-net/simplenet-upd.c \
 	src/networks/model-net/torus.c \
+	src/networks/model-net/express-mesh.C \
 	src/networks/model-net/dragonfly.c \
 	src/networks/model-net/dragonfly-custom.C \
 	src/networks/model-net/slimfly.c \

src/iokernellang/codeslexer.c

-#line 2 "codeslexer.c"
+#line 2 "../src/iokernellang/codeslexer.c"
 
-#line 4 "codeslexer.c"
+#line 4 "../src/iokernellang/codeslexer.c"
 
 #define  YY_INT_ALIGNED short int
 
@@ -54,6 +54,7 @@ typedef int flex_int32_t;
 typedef unsigned char flex_uint8_t; 
 typedef unsigned short int flex_uint16_t;
 typedef unsigned int flex_uint32_t;
+#endif /* ! C99 */
 
 /* Limits of integral types. */
 #ifndef INT8_MIN
@@ -84,8 +85,6 @@ typedef unsigned int flex_uint32_t;
 #define UINT32_MAX             (4294967295U)
 #endif
 
-#endif /* ! C99 */
-
 #endif /* ! FLEXINT_H */
 
 #ifdef __cplusplus
@@ -159,15 +158,7 @@ typedef void* yyscan_t;
 
 /* Size of default input buffer. */
 #ifndef YY_BUF_SIZE
-#ifdef __ia64__
-/* On IA-64, the buffer size is 16k, not 8k.
- * Moreover, YY_BUF_SIZE is 2*YY_READ_BUF_SIZE in the general case.
- * Ditto for the __ia64__ case accordingly.
- */
-#define YY_BUF_SIZE 32768
-#else
 #define YY_BUF_SIZE 16384
-#endif /* __ia64__ */
 #endif
 
 /* The state buf must be large enough to hold one state per character in the main buffer.
@@ -538,20 +529,20 @@ static yyconst flex_int32_t yy_rule_can_match_eol[37] =
 #define yymore() yymore_used_but_not_detected
 #define YY_MORE_ADJ 0
 #define YY_RESTORE_YY_MORE_OFFSET
-#line 1 "codeslexer.l"
+#line 1 "../src/iokernellang/codeslexer.l"
 /*
  * Copyright (C) 2013 University of Chicago.
  * See COPYRIGHT notice in top-level directory.
  *
  */
-#line 8 "codeslexer.l"
+#line 8 "../src/iokernellang/codeslexer.l"
 #include <stdlib.h>
-#include "CodesIOKernelTypes.h"
-#include "CodesIOKernelParser.h"
+#include "src/iokernellang/CodesIOKernelTypes.h"
+#include "src/iokernellang/CodesIOKernelParser.h"
 
 #define YY_NO_INPUT
 
-#line 25 "codeslexer.l"
+#line 25 "../src/iokernellang/codeslexer.l"
 	YYLTYPE *CodesIOKernel_get_lloc  (yyscan_t yyscanner);
 	int CodesIOKernel_lex_init (yyscan_t* scanner);
 	int CodesIOKernel_lex(YYSTYPE * lvalp, YYLTYPE * llocp, void * scanner);
@@ -568,7 +559,7 @@ static yyconst flex_int32_t yy_rule_can_match_eol[37] =
         #define YY_USER_ACTION /* no user action */;
 #endif
 
-#line 572 "codeslexer.c"
+#line 563 "../src/iokernellang/codeslexer.c"
 
 #define INITIAL 0
 
@@ -703,12 +694,7 @@ static int input (yyscan_t yyscanner );
 
 /* Amount of stuff to slurp up with each read. */
 #ifndef YY_READ_BUF_SIZE
-#ifdef __ia64__
-/* On IA-64, the buffer size is 16k, not 8k */
-#define YY_READ_BUF_SIZE 16384
-#else
 #define YY_READ_BUF_SIZE 8192
-#endif /* __ia64__ */
 #endif
 
 /* Copy whatever the last rule matched to the standard output. */
@@ -727,7 +713,7 @@ static int input (yyscan_t yyscanner );
 	if ( YY_CURRENT_BUFFER_LVALUE->yy_is_interactive ) \
 		{ \
 		int c = '*'; \
-		size_t n; \
+		unsigned n; \
 		for ( n = 0; n < max_size && \
 			     (c = getc( yyin )) != EOF && c != '\n'; ++n ) \
 			buf[n] = (char) c; \
@@ -812,10 +798,10 @@ YY_DECL
 	register int yy_act;
     struct yyguts_t * yyg = (struct yyguts_t*)yyscanner;
 
-#line 43 "codeslexer.l"
+#line 43 "../src/iokernellang/codeslexer.l"
 
 
-#line 819 "codeslexer.c"
+#line 805 "../src/iokernellang/codeslexer.c"
 
     yylval = yylval_param;
 
@@ -916,7 +902,7 @@ do_action:	/* This label is used only to access EOF actions. */
 
 case 1:
 YY_RULE_SETUP
-#line 45 "codeslexer.l"
+#line 45 "../src/iokernellang/codeslexer.l"
 {
 		if(yyextra->text != NULL)
                 {
@@ -929,7 +915,7 @@ YY_RULE_SETUP
 	YY_BREAK
 case 2:
 YY_RULE_SETUP
-#line 55 "codeslexer.l"
+#line 55 "../src/iokernellang/codeslexer.l"
 {
 		if(yyextra->text != NULL)
                 {
@@ -942,7 +928,7 @@ YY_RULE_SETUP
 	YY_BREAK
 case 3:
 YY_RULE_SETUP
-#line 65 "codeslexer.l"
+#line 65 "../src/iokernellang/codeslexer.l"
 {
 		if(yyextra->text != NULL)
                 {
@@ -955,7 +941,7 @@ YY_RULE_SETUP
 	YY_BREAK
 case 4:
 YY_RULE_SETUP
-#line 75 "codeslexer.l"
+#line 75 "../src/iokernellang/codeslexer.l"
 {
 		if(yyextra->text != NULL)
                 {
@@ -967,166 +953,166 @@ YY_RULE_SETUP
 	YY_BREAK
 case 5:
 YY_RULE_SETUP
-#line 84 "codeslexer.l"
+#line 84 "../src/iokernellang/codeslexer.l"
 return GE;
 	YY_BREAK
 case 6:
 YY_RULE_SETUP
-#line 85 "codeslexer.l"
+#line 85 "../src/iokernellang/codeslexer.l"
 return LE;
 	YY_BREAK
 case 7:
 YY_RULE_SETUP
-#line 86 "codeslexer.l"
+#line 86 "../src/iokernellang/codeslexer.l"
 return EQ;
 	YY_BREAK
 case 8:
 YY_RULE_SETUP
-#line 87 "codeslexer.l"
+#line 87 "../src/iokernellang/codeslexer.l"
 return NE;
 	YY_BREAK
 case 9:
 YY_RULE_SETUP
-#line 88 "codeslexer.l"
+#line 88 "../src/iokernellang/codeslexer.l"
 return WHILE;
 	YY_BREAK
 case 10:
 YY_RULE_SETUP
-#line 89 "codeslexer.l"
+#line 89 "../src/iokernellang/codeslexer.l"
 return IF;
 	YY_BREAK
 case 11:
 YY_RULE_SETUP
-#line 90 "codeslexer.l"
+#line 90 "../src/iokernellang/codeslexer.l"
 return ELSE;
 	YY_BREAK
 case 12:
 YY_RULE_SETUP
-#line 91 "codeslexer.l"
+#line 91 "../src/iokernellang/codeslexer.l"
 return PRINT;
 	YY_BREAK
 case 13:
 YY_RULE_SETUP
-#line 92 "codeslexer.l"
+#line 92 "../src/iokernellang/codeslexer.l"
 return WRITE;
 	YY_BREAK
 case 14:
 YY_RULE_SETUP
-#line 93 "codeslexer.l"
+#line 93 "../src/iokernellang/codeslexer.l"
 return WRITEAT;
 	YY_BREAK
 case 15:
 YY_RULE_SETUP
-#line 94 "codeslexer.l"
+#line 94 "../src/iokernellang/codeslexer.l"
 return WRITE_ALL;
 	YY_BREAK
 case 16:
 YY_RULE_SETUP
-#line 95 "codeslexer.l"
+#line 95 "../src/iokernellang/codeslexer.l"
 return WRITEAT_ALL;
 	YY_BREAK
 case 17:
 YY_RULE_SETUP
-#line 96 "codeslexer.l"
+#line 96 "../src/iokernellang/codeslexer.l"
 return READ;
 	YY_BREAK
 case 18:
 YY_RULE_SETUP
-#line 97 "codeslexer.l"
+#line 97 "../src/iokernellang/codeslexer.l"
 return READAT;
 	YY_BREAK
 case 19:
 YY_RULE_SETUP
-#line 98 "codeslexer.l"
+#line 98 "../src/iokernellang/codeslexer.l"
 return READ_ALL;
 	YY_BREAK
 case 20:
 YY_RULE_SETUP
-#line 99 "codeslexer.l"
+#line 99 "../src/iokernellang/codeslexer.l"
 return READAT_ALL;
 	YY_BREAK
 case 21:
 YY_RULE_SETUP
-#line 100 "codeslexer.l"
+#line 100 "../src/iokernellang/codeslexer.l"
 return SYNC;
 	YY_BREAK
 case 22:
 YY_RULE_SETUP
-#line 101 "codeslexer.l"
+#line 101 "../src/iokernellang/codeslexer.l"
 return DELETE;
 	YY_BREAK
 case 23:
 YY_RULE_SETUP
-#line 102 "codeslexer.l"
+#line 102 "../src/iokernellang/codeslexer.l"
 return SLEEP;
 	YY_BREAK
 case 24:
 YY_RULE_SETUP
-#line 103 "codeslexer.l"
+#line 103 "../src/iokernellang/codeslexer.l"
 return OPEN;
 	YY_BREAK
 case 25:
 YY_RULE_SETUP
-#line 104 "codeslexer.l"
+#line 104 "../src/iokernellang/codeslexer.l"
 return CLOSE;
 	YY_BREAK
 case 26:
 YY_RULE_SETUP
-#line 105 "codeslexer.l"
+#line 105 "../src/iokernellang/codeslexer.l"
 return FLUSH;
 	YY_BREAK
 case 27:
 YY_RULE_SETUP
-#line 106 "codeslexer.l"
+#line 106 "../src/iokernellang/codeslexer.l"
 return SEEK;
 	YY_BREAK
 case 28:
 YY_RULE_SETUP
-#line 107 "codeslexer.l"
+#line 107 "../src/iokernellang/codeslexer.l"
 return GETGROUPID;
 	YY_BREAK
 case 29:
 YY_RULE_SETUP
-#line 108 "codeslexer.l"
+#line 108 "../src/iokernellang/codeslexer.l"
 return GETNUMGROUPS;
 	YY_BREAK
 case 30:
 YY_RULE_SETUP
-#line 109 "codeslexer.l"
+#line 109 "../src/iokernellang/codeslexer.l"
 return GETGROUPRANK;
 	YY_BREAK
 case 31:
 YY_RULE_SETUP
-#line 110 "codeslexer.l"
+#line 110 "../src/iokernellang/codeslexer.l"
 return GETGROUPSIZE;
 	YY_BREAK
 case 32:
 YY_RULE_SETUP
-#line 111 "codeslexer.l"
+#line 111 "../src/iokernellang/codeslexer.l"
 return GETCURTIME;
 	YY_BREAK
 case 33:
 YY_RULE_SETUP
-#line 112 "codeslexer.l"
+#line 112 "../src/iokernellang/codeslexer.l"
 return EXIT;
 	YY_BREAK
 case 34:
 /* rule 34 can match eol */
 YY_RULE_SETUP
-#line 114 "codeslexer.l"
+#line 114 "../src/iokernellang/codeslexer.l"
 ;       /* ignore whitespace */
 	YY_BREAK
 case 35:
 YY_RULE_SETUP
-#line 116 "codeslexer.l"
+#line 116 "../src/iokernellang/codeslexer.l"
 fprintf(stderr, "Unknown character\n");
 	YY_BREAK
 case 36:
 YY_RULE_SETUP
-#line 117 "codeslexer.l"
+#line 117 "../src/iokernellang/codeslexer.l"
 ECHO;
 	YY_BREAK
-#line 1130 "codeslexer.c"
+#line 1116 "../src/iokernellang/codeslexer.c"
 case YY_STATE_EOF(INITIAL):
 	yyterminate();
 
@@ -1867,8 +1853,8 @@ YY_BUFFER_STATE CodesIOKernel__scan_string (yyconst char * yystr , yyscan_t yysc
 
 /** Setup the input buffer state to scan the given bytes. The next call to CodesIOKernel_lex() will
  * scan from a @e copy of @a bytes.
- * @param yybytes the byte buffer to scan
- * @param _yybytes_len the number of bytes in the buffer pointed to by @a bytes.
+ * @param bytes the byte buffer to scan
+ * @param len the number of bytes in the buffer pointed to by @a bytes.
  * @param yyscanner The scanner object.
  * @return the newly allocated buffer state object.
  */
@@ -2274,7 +2260,7 @@ void CodesIOKernel_free (void * ptr , yyscan_t yyscanner)
 
 #define YYTABLES_NAME "yytables"
 
-#line 117 "codeslexer.l"
+#line 117 "../src/iokernellang/codeslexer.l"
 
 
 

src/iokernellang/codeslexer.h

@@ -2,9 +2,9 @@
 #define CodesIOKernel_HEADER_H 1
 #define CodesIOKernel_IN_HEADER 1
 
-#line 6 "codeslexer.h"
+#line 6 "../src/iokernellang/codeslexer.h"
 
-#line 8 "codeslexer.h"
+#line 8 "../src/iokernellang/codeslexer.h"
 
 #define  YY_INT_ALIGNED short int
 
@@ -58,6 +58,7 @@ typedef int flex_int32_t;
 typedef unsigned char flex_uint8_t; 
 typedef unsigned short int flex_uint16_t;
 typedef unsigned int flex_uint32_t;
+#endif /* ! C99 */
 
 /* Limits of integral types. */
 #ifndef INT8_MIN
@@ -88,8 +89,6 @@ typedef unsigned int flex_uint32_t;
 #define UINT32_MAX             (4294967295U)
 #endif
 
-#endif /* ! C99 */
-
 #endif /* ! FLEXINT_H */
 
 #ifdef __cplusplus
@@ -132,15 +131,7 @@ typedef void* yyscan_t;
 
 /* Size of default input buffer. */
 #ifndef YY_BUF_SIZE
-#ifdef __ia64__
-/* On IA-64, the buffer size is 16k, not 8k.
- * Moreover, YY_BUF_SIZE is 2*YY_READ_BUF_SIZE in the general case.
- * Ditto for the __ia64__ case accordingly.
- */
-#define YY_BUF_SIZE 32768
-#else
 #define YY_BUF_SIZE 16384
-#endif /* __ia64__ */
 #endif
 
 #ifndef YY_TYPEDEF_YY_BUFFER_STATE
@@ -311,12 +302,7 @@ static int yy_flex_strlen (yyconst char * ,yyscan_t yyscanner);
 
 /* Amount of stuff to slurp up with each read. */
 #ifndef YY_READ_BUF_SIZE
-#ifdef __ia64__
-/* On IA-64, the buffer size is 16k, not 8k */
-#define YY_READ_BUF_SIZE 16384
-#else
 #define YY_READ_BUF_SIZE 8192
-#endif /* __ia64__ */
 #endif
 
 /* Number of entries by which start-condition stack grows. */
@@ -351,9 +337,9 @@ extern int CodesIOKernel_lex \
 #undef YY_DECL
 #endif
 
-#line 117 "codeslexer.l"
+#line 117 "../src/iokernellang/codeslexer.l"
 
 
-#line 358 "codeslexer.h"
+#line 344 "../src/iokernellang/codeslexer.h"
 #undef CodesIOKernel_IN_HEADER
 #endif /* CodesIOKernel_HEADER_H */

src/networks/model-net/common-net.c

+#include "codes/net/common-net.h"
+#include "codes/quickhash.h"
+#include "assert.h"
+
+void append_to_message_list(  
+        message_list ** thisq,
+        message_list ** thistail,
+        int index, 
+        message_list *msg) {
+    if(thisq[index] == NULL) {
+        thisq[index] = msg;
+    } else {
+        thistail[index]->next = msg;
+        msg->prev = thistail[index];
+    } 
+    thistail[index] = msg;
+}
+
+void prepend_to_message_list(  
+        message_list ** thisq,
+        message_list ** thistail,
+        int index, 
+        message_list *msg) {
+    if(thisq[index] == NULL) {
+        thistail[index] = msg;
+    } else {
+        thisq[index]->prev = msg;
+        msg->next = thisq[index];
+    } 
+    thisq[index] = msg;
+}
+
+message_list* return_head(
+        message_list ** thisq,
+        message_list ** thistail,
+        int index) {
+    message_list *head = thisq[index];
+    if(head != NULL) {
+        thisq[index] = head->next;
+        if(head->next != NULL) {
+            head->next->prev = NULL;
+            head->next = NULL;
+        } else {
+            thistail[index] = NULL;
+        }
+    }
+    return head;
+}
+
+message_list* return_tail(
+        message_list ** thisq,
+        message_list ** thistail,
+        int index) {
+    message_list *tail = thistail[index];
+    assert(tail);
+    if(tail->prev != NULL) {
+        tail->prev->next = NULL;
+        thistail[index] = tail->prev;
+        tail->prev = NULL;
+    } else {
+        thistail[index] = NULL;
+        thisq[index] = NULL;
+    }