Process OF messages spanning multiple TCP segments

ZodiacFX/src/openflow/openflow.c

@@ -1,4 +1,6 @@
 /**
+ * vi: ts=4:sw=4
+ *
  * @file
  * openflow.c
  *
@@ -73,15 +75,20 @@ int tcp_con_state = -1;
 int tcp_wait = 0;
 int totaltime = 0;
 int heartbeat = 0;
-int multi_pos;
 uint32_t reply_more_xid = 0;
 bool reply_more_flag = false;
 bool rcv_freq;
 
+// Buffer for multi-segment messages
+#define PACKET_BUFFER_SIZE		(2*TCP_MSS+64)	// TDDO: Ideally would be (2*1536)
+static uint8_t packet_buffer[PACKET_BUFFER_SIZE];
+static unsigned int packet_buffer_off = 0;
+static unsigned int packet_buffer_len = 0;
+
 // Internal Functions
 void OF_hello(void);
 void echo_request(void);
-void echo_reply(struct ofp_header *ofph, int size, int len);
+void echo_reply(struct ofp_header *ofph, int len);
 err_t TCPready(void *arg, struct tcp_pcb *tpcb, err_t err);
 void tcp_error(void * arg, err_t err);
 static err_t of_receive(void *arg, struct tcp_pcb *tpcb, struct pbuf *p, err_t err);
@@ -133,43 +140,44 @@ void nnOF_tablelookup(uint8_t *p_uc_data, uint32_t *ul_size, int port)
 */
 static err_t of_receive(void *arg, struct tcp_pcb *tpcb, struct pbuf *p, err_t err)
 {
-	char *pc;
-	char packetbuffer[1536];
 	heartbeat = 0;	// Reset heartbeat counter
 
 	if (err == ERR_OK && p != NULL)
 	{
-		tcp_recved(tpcb, p->tot_len);
-        pc=(char *)p->payload;	//pointer to the payload
-        int len = p->tot_len;	//size of the payload
-        for (int i=0; i<len; i++)packetbuffer[i] = pc[i];	//copy to our own buffer
-        pbuf_free(p);	//Free the packet buffer
-		TRACE("openflow.c: OpenFlow data received (%d bytes)", len);
-		struct ofp_header *ofph;
-		int size = 0;
-		int plen = 0;
-
-		while (size < len)
+		int plen = p->tot_len;	//size of the payload
+		if (packet_buffer_len + plen > PACKET_BUFFER_SIZE) {
+			TRACE("openflow.c: packet buffer exceeded, aborting!!!");
+			return ERR_ABRT;
+		}
+		memcpy(&packet_buffer[packet_buffer_len], p->payload, plen);	// append to our own buffer
+		packet_buffer_len += plen;
+		TRACE("openflow.c: OpenFlow data received (%d bytes)", plen);
+
+		// Accept the TCP data and release the packet
+		tcp_recved(tpcb, plen);
+		pbuf_free(p);
+
+		while (packet_buffer_off < packet_buffer_len)
 		{
-			ofph = &packetbuffer[size];
-			if (size == 0) multi_pos = 0;
+			struct ofp_header *ofph = &packet_buffer[packet_buffer_off];
 			if (ofph->length == 0 || ofph->version == 0){
-				return ERR_OK;	//Not an OpenFlow packet
+				TRACE("openflow.c: Invalid OpenFlow packet, aborting!");
+				return ERR_ABRT;
 			}
-			plen = htons(ofph->length);
-
-			if (ofph->version > 6 || ofph->type > 30) //	Invalid OpenFlow message
+			if (ofph->version > 6 || ofph->type > 30)
 			{
-				TRACE("openflow.c: Invalid OpenFlow command, ignoring!");
-				return ERR_OK;
+				TRACE("openflow.c: Invalid OpenFlow packet values, aborting!");
+				return ERR_ABRT;
 			}
-			size = size + plen;
-			if (size > len) // corrupt OpenFlow command
+
+			int ofp_len = htons(ofph->length);
+			if ((packet_buffer_off + ofp_len) > packet_buffer_len)
 			{
+				// TRACE("openflow.c: Partial OpenFlow message - waiting for more data...");
 				break;
-				TRACE("openflow.c: Corrupt OpenFlow Message!!!");	
 			}
-			TRACE("openflow.c: Processing %d byte OpenFlow message %u (%d)", plen, htonl(ofph->xid), size);
+			packet_buffer_off += ofp_len;
+			TRACE("openflow.c: Processing %d byte OpenFlow message %u", ofp_len, htonl(ofph->xid));
 
 			switch(ofph->type)
 			{
@@ -190,14 +198,24 @@ static err_t of_receive(void *arg, struct tcp_pcb *tpcb, struct pbuf *p, err_t e
 				break;
 
 				case OFPT10_ECHO_REQUEST:
-					echo_reply(ofph, size, len);
+					echo_reply(ofph, ofp_len);
 				break;
 
 				default:
-					if (OF_Version == 0x01) of10_message(ofph, size, len);
-					if (OF_Version == 0x04) of13_message(ofph, size, len);
+					if (OF_Version == 0x01) of10_message(ofph, ofp_len);
+					if (OF_Version == 0x04) of13_message(ofph, ofp_len);
 			};
+		}
 
+		if (packet_buffer_off == packet_buffer_len) {
+			packet_buffer_off = 0;
+			packet_buffer_len = 0;
+		} else {
+			unsigned int rem = packet_buffer_len - packet_buffer_off;
+			memcpy(packet_buffer, &packet_buffer[packet_buffer_off], rem);
+			packet_buffer_off = 0;
+			packet_buffer_len = rem;
+			TRACE("openflow.c: Partial OpenFlow message - keeping %d bytes", rem);
 		}
 	} else {
 		pbuf_free(p);
@@ -259,7 +277,7 @@ void OF_hello(void)
 *	@param xid - transaction ID
 *
 */
-void echo_reply(struct ofp_header *ofph, int size, int len)
+void echo_reply(struct ofp_header *ofph, int len)
 {
 	// Change the message type to Echo Reply and return any data that was sent
 	ofph->type   = OFPT10_ECHO_REPLY;
@@ -405,6 +423,8 @@ void task_openflow(void)
 */
 err_t TCPready(void *arg, struct tcp_pcb *tpcb, err_t err)
 {
+	packet_buffer_off = 0;
+	packet_buffer_len = 0;
 	tcp_con_state = true;
 	tcp_recv(tpcb, of_receive);
 	tcp_poll(tpcb, NULL, 4);

ZodiacFX/src/openflow/openflow.h

@@ -134,8 +134,8 @@ void task_openflow(void);
 void nnOF_tablelookup(uint8_t *p_uc_data, uint32_t *ul_size, int port);
 void nnOF10_tablelookup(uint8_t *p_uc_data, uint32_t *ul_size, int port);
 void nnOF13_tablelookup(uint8_t *p_uc_data, uint32_t *ul_size, int port);
-void of10_message(struct ofp_header *ofph, int size, int len);
-void of13_message(struct ofp_header *ofph, int size, int len);
+void of10_message(struct ofp_header *ofph, int len);
+void of13_message(struct ofp_header *ofph, int len);
 void multi_flow_more_reply13(void);
 void barrier10_reply(uint32_t xid);
 void barrier13_reply(uint32_t xid);

ZodiacFX/src/openflow/openflow_10.c

@@ -339,7 +339,7 @@ void nnOF10_tablelookup(uint8_t *p_uc_data, uint32_t *ul_size, int port)
 	return;	// Should only get to here if the action is unknown
 }
 
-void of10_message(struct ofp_header *ofph, int size, int len)
+void of10_message(struct ofp_header *ofph, int len)
 {
 	struct ofp_stats_request *stats_req;
 	switch(ofph->type)

ZodiacFX/src/openflow/openflow_13.c

@@ -65,7 +65,6 @@ extern struct table_counter table_counters[MAX_TABLES];
 extern uint8_t port_status[TOTAL_PORTS];
 extern struct ofp_switch_config Switch_config;
 extern uint8_t shared_buffer[SHARED_BUFFER_LEN];
-extern int multi_pos;
 extern uint8_t NativePortMatrix;
 extern bool reply_more_flag;
 extern uint32_t reply_more_xid;
@@ -663,9 +662,11 @@ void nnOF13_tablelookup(uint8_t *p_uc_data, uint32_t *ul_size, int port)
 	return;
 }
 
-void of13_message(struct ofp_header *ofph, int size, int len)
+void of13_message(struct ofp_header *ofph, int len)
 {
 	struct ofp13_multipart_request *multi_req;
+	int multi_len;
+
 	TRACE("openflow_13.c: %u: OpenFlow message received type = %d", htonl(ofph->xid), ofph->type);
 	switch(ofph->type)
 	{
@@ -697,73 +698,84 @@ void of13_message(struct ofp_header *ofph, int size, int len)
 
 		case OFPT13_MULTIPART_REQUEST:
 		multi_req  = (struct ofp13_multipart_request *) ofph;
+		if(multi_req->flags != 0)
+		{
+			TRACE("openflow_13.c: unsupported MULTIPART 'flags' request: %04x", multi_req->flags);
+			return;
+		}
+
 		if ( ntohs(multi_req->type) == OFPMP13_DESC )
 		{
-			multi_pos += multi_desc_reply13(&shared_buffer[multi_pos], multi_req);
+			multi_len = multi_desc_reply13(shared_buffer, multi_req);
 		}
 
 		if ( ntohs(multi_req->type) == 	OFPMP13_FLOW )
 		{
-			multi_pos += multi_flow_reply13(&shared_buffer[multi_pos], multi_req);
+			multi_len = multi_flow_reply13(shared_buffer, multi_req);
 		}
 
 		if ( ntohs(multi_req->type) == OFPMP13_AGGREGATE )
 		{
-			multi_pos += multi_aggregate_reply13(&shared_buffer[multi_pos], multi_req);
+			multi_len = multi_aggregate_reply13(shared_buffer, multi_req);
 		}
 
 		if ( ntohs(multi_req->type) == OFPMP13_PORT_STATS )
 		{
-			multi_pos += multi_portstats_reply13(&shared_buffer[multi_pos], multi_req);
+			multi_len = multi_portstats_reply13(shared_buffer, multi_req);
 		}
 
 		if ( ntohs(multi_req->type) == OFPMP13_PORT_DESC )
 		{
-			multi_pos += multi_portdesc_reply13(&shared_buffer[multi_pos], multi_req);
+			multi_len = multi_portdesc_reply13(shared_buffer, multi_req);
 		}
 
 		if ( ntohs(multi_req->type) == OFPMP13_METER )
 		{
-			multi_pos += multi_meter_stats_reply13(&shared_buffer[multi_pos], multi_req);
+			multi_len = multi_meter_stats_reply13(shared_buffer, multi_req);
 		}
 
 		if ( ntohs(multi_req->type) == OFPMP13_METER_CONFIG )
 		{
-			multi_pos += multi_meter_config_reply13(&shared_buffer[multi_pos], multi_req);
+			multi_len = multi_meter_config_reply13(shared_buffer, multi_req);
 		}
 
 		if ( ntohs(multi_req->type) == OFPMP13_METER_FEATURES )
 		{
-			multi_pos += multi_meter_features_reply13(&shared_buffer[multi_pos], multi_req);
+			multi_len = multi_meter_features_reply13(shared_buffer, multi_req);
 		}
 
 		if ( ntohs(multi_req->type) == OFPMP13_GROUP_FEATURES )
 		{
-			multi_pos += multi_group_features_reply13(&shared_buffer[multi_pos], multi_req);
+			multi_len = multi_group_features_reply13(shared_buffer, multi_req);
 		}
 
 		if ( ntohs(multi_req->type) == OFPMP13_GROUP_DESC )
 		{
-			multi_pos += multi_group_desc_reply13(&shared_buffer[multi_pos], multi_req);
+			multi_len = multi_group_desc_reply13(shared_buffer, multi_req);
 		}
 
 		if ( ntohs(multi_req->type) == OFPMP13_GROUP )
 		{
-			multi_pos += multi_group_stats_reply13(&shared_buffer[multi_pos], multi_req);
+			multi_len = multi_group_stats_reply13(shared_buffer, multi_req);
 		}
 
 		if ( htons(multi_req->type) == OFPMP13_TABLE_FEATURES )
 		{
 			/**** Floodlight v1.2 crashes when it gets this reply, removed for the moment. *****/
-			multi_pos += multi_tablefeat_reply13(&shared_buffer[multi_pos], multi_req);
+			multi_len = multi_tablefeat_reply13(shared_buffer, multi_req);
 			//of_error13(ofph, OFPET13_BAD_REQUEST, OFPBRC13_BAD_TYPE);
 		}
 
 		if ( ntohs(multi_req->type) == OFPMP13_TABLE )
 		{
-			multi_pos += multi_table_reply13(&shared_buffer[multi_pos], multi_req);
+			multi_len = multi_table_reply13(shared_buffer, multi_req);
 		}
 
+
+		if (multi_len !=0)
+		{
+			sendtcp(shared_buffer, multi_len, 0);
+		}
 		break;
 
 		case OFPT10_PACKET_OUT:
@@ -778,11 +790,6 @@ void of13_message(struct ofp_header *ofph, int size, int len)
 		meter_mod13(ofph);
 		break;
 	};
-
-	if (size == len && multi_pos !=0)
-	{
-		sendtcp(&shared_buffer, multi_pos, 0);
-	}
 	return;
 }
 
@@ -1091,8 +1098,9 @@ int multi_portdesc_reply13(uint8_t *buffer, struct ofp13_multipart_request *msg)
 int multi_table_reply13(uint8_t *buffer, struct ofp13_multipart_request *msg)
 {
 	int len = offsetof(struct ofp13_multipart_reply, body) + sizeof(struct ofp13_table_stats) * MAX_TABLES;
-	if (SHARED_BUFFER_LEN - multi_pos < len){
-			return 0; // guard for buffer overrun
+	if (SHARED_BUFFER_LEN < len) { // guard for buffer overrun
+		TRACE("openflow_13.c: multi-table reply space exceeded, ignoring");
+		return 0;
 	}
 	bzero(buffer, len);
 	struct ofp13_multipart_reply *reply = buffer;