1M HTTP rps على وحدة معالجة مركزية واحدة. DPDK بدلاً من nginx + linux kernel TCP / IP

 tatic void process_tcp(struct rte_mbuf* m, struct tcp_hdr* tcp_header, struct tcp_key* key, void* data, size_t data_size) { TRACE; struct tcp_state* state; if (rte_hash_lookup_data(g_clients, key, (void**)&state) < 0) //Documentaion lies!!! { TRACE; if ((tcp_header->tcp_flags & 0x2) != 0) // SYN { TRACE; struct ether_hdr* eth_header = rte_pktmbuf_mtod(m, struct ether_hdr*); if (rte_mempool_get(g_tcp_state_pool, (void**)&state) < 0) { ERROR("tcp state alloc fail"); return; } memcpy(&state->tcp_template, &g_tcp_packet_template, sizeof(g_tcp_packet_template)); memcpy(&state->tcp_template.eth.d_addr, &eth_header->s_addr, 6); state->tcp_template.ip.dst_addr = key->ip; state->tcp_template.tcp.dst_port = key->port; state->remote_seq = htonl(tcp_header->sent_seq); #pragma GCC diagnostic push #pragma GCC diagnostic ignored "-Wpointer-to-int-cast" state->my_seq_start = (uint32_t)state; // not very secure. #pragma GCC diagnostic pop state->fin_sent = 0; state->http.state = HTTP_START; state->http.request_url_size = 0; //not thread safe! only one core used if (rte_hash_add_key_data(g_clients, key, state) == 0) { struct tcp_hdr* new_tcp_header; struct rte_mbuf* packet = build_packet(state, 12, &new_tcp_header); if (packet != NULL) { new_tcp_header->rx_win = TX_WINDOW_SIZE; new_tcp_header->sent_seq = htonl(state->my_seq_start); state->my_seq_sent = state->my_seq_start+1; ++state->remote_seq; new_tcp_header->recv_ack = htonl(state->remote_seq); new_tcp_header->tcp_flags = 0x12; // mss = 1380, no window scaling uint8_t options[12] = {0x02, 0x04, 0x05, 0x64, 0x03, 0x03, 0x00, 0x01, 0x01, 0x01, 0x01, 0x01}; memcpy((uint8_t*)new_tcp_header + sizeof(struct tcp_hdr), options, 12); new_tcp_header->data_off = 0x80; send_packet(new_tcp_header, packet); } else { ERROR("rte_pktmbuf_alloc, tcp synack"); } } else { ERROR("can't add connection to table"); rte_mempool_put(g_tcp_state_pool, state); } } else { ERROR("lost connection"); } return; } if ((tcp_header->tcp_flags & 0x2) != 0) // SYN retransmit { //not thread safe! only one core used if (rte_hash_del_key(g_clients, key) < 0) { ERROR("can't delete key"); } else { rte_mempool_put(g_tcp_state_pool, state); return process_tcp(m, tcp_header, key, data, data_size); } } if ((tcp_header->tcp_flags & 0x10) != 0) // ACK { TRACE; uint32_t ack_delta = htonl(tcp_header->recv_ack) - state->my_seq_start; uint32_t my_max_ack_delta = state->my_seq_sent - state->my_seq_start; if (ack_delta == 0) { if ((data_size == 0) && (tcp_header->tcp_flags == 0x10)) { ERROR("need to retransmit. not supported"); } } else if (ack_delta <= my_max_ack_delta) { state->my_seq_start += ack_delta; } else { ERROR("ack on unsent seq"); } } if (data_size > 0) { TRACE; uint32_t packet_seq = htonl(tcp_header->sent_seq); if (state->remote_seq == packet_seq) { feed_http(data, data_size, state); state->remote_seq += data_size; } else if (state->remote_seq-1 == packet_seq) // keepalive { struct tcp_hdr* new_tcp_header; struct rte_mbuf* packet = build_packet(state, 0, &new_tcp_header); if (packet != NULL) { new_tcp_header->rx_win = TX_WINDOW_SIZE; new_tcp_header->sent_seq = htonl(state->my_seq_sent); new_tcp_header->recv_ack = htonl(state->remote_seq); send_packet(new_tcp_header, packet); } else { ERROR("rte_pktmbuf_alloc, tcp ack keepalive"); } } else { struct tcp_hdr* new_tcp_header; struct rte_mbuf* packet = build_packet(state, state->http.last_message_size, &new_tcp_header); TRACE; if (packet != NULL) { new_tcp_header->rx_win = TX_WINDOW_SIZE; new_tcp_header->sent_seq = htonl(state->my_seq_sent - state->http.last_message_size); new_tcp_header->recv_ack = htonl(state->remote_seq); memcpy((char*)new_tcp_header+sizeof(struct tcp_hdr), &state->http.last_message, state->http.last_message_size); send_packet(new_tcp_header, packet); } else { ERROR("rte_pktmbuf_alloc, tcp fin ack"); } //ERROR("my bad tcp stack implementation((("); } } if ((tcp_header->tcp_flags & 0x04) != 0) // RST { TRACE; //not thread safe! only one core used if (rte_hash_del_key(g_clients, key) < 0) { ERROR("can't delete key"); } else { rte_mempool_put(g_tcp_state_pool, state); } } else if ((tcp_header->tcp_flags & 0x01) != 0) // FIN { struct tcp_hdr* new_tcp_header; struct rte_mbuf* packet = build_packet(state, 0, &new_tcp_header); TRACE; if (packet != NULL) { new_tcp_header->rx_win = TX_WINDOW_SIZE; new_tcp_header->sent_seq = htonl(state->my_seq_sent); new_tcp_header->recv_ack = htonl(state->remote_seq + 1); if (!state->fin_sent) { TRACE; new_tcp_header->tcp_flags = 0x11; // !@#$ the last ack } send_packet(new_tcp_header, packet); } else { ERROR("rte_pktmbuf_alloc, tcp fin ack"); } //not thread safe! only one core used if (rte_hash_del_key(g_clients, key) < 0) { ERROR("can't delete key"); } else { rte_mempool_put(g_tcp_state_pool, state); } } } 

 static void feed_http(void* data, size_t data_size, struct tcp_state* state) { TRACE; size_t remaining_data = data_size; char* current = (char*)data; struct http_state* http = &state->http; if (http->state == HTTP_BAD_STATE) { TRACE; return; } while (remaining_data > 0) { switch(http->state) { case HTTP_START: { if (*current == 'G') { http->state = HTTP_READ_G; } else { http->state = HTTP_BAD_STATE; } break; } case HTTP_READ_G: { if (*current == 'E') { http->state = HTTP_READ_E; } else { http->state = HTTP_BAD_STATE; } break; } case HTTP_READ_E: { if (*current == 'T') { http->state = HTTP_READ_T; } else { http->state = HTTP_BAD_STATE; } break; } case HTTP_READ_T: { if (*current == ' ') { http->state = HTTP_READ_SPACE; } else { http->state = HTTP_BAD_STATE; } break; } case HTTP_READ_SPACE: { if (*current != ' ') { http->request_url[http->request_url_size] = *current; ++http->request_url_size; if (http->request_url_size > MAX_URL_SIZE) { http->state = HTTP_BAD_STATE; } } else { http->state = HTTP_READ_URL; http->request_url[http->request_url_size] = '\0'; } break; } case HTTP_READ_URL: { if (*current == '\r') { http->state = HTTP_READ_R1; } break; } case HTTP_READ_R1: { if (*current == '\n') { http->state = HTTP_READ_N1; } else if (*current == '\r') { http->state = HTTP_READ_R1; } else { http->state = HTTP_READ_URL; } break; } case HTTP_READ_N1: { if (*current == '\r') { http->state = HTTP_READ_R2; } else { http->state = HTTP_READ_URL; } break; } case HTTP_READ_R2: { if (*current == '\n') { TRACE; char content_length[32]; sprintf(content_length, "%lu", g_http_part2_size - 4 + http->request_url_size + g_http_part3_size); size_t content_length_size = strlen(content_length); size_t total_data_size = g_http_part1_size + g_http_part2_size + g_http_part3_size + http->request_url_size + content_length_size; struct tcp_hdr* tcp_header; struct rte_mbuf* packet = build_packet(state, total_data_size, &tcp_header); if (packet != NULL) { tcp_header->rx_win = TX_WINDOW_SIZE; tcp_header->sent_seq = htonl(state->my_seq_sent); tcp_header->recv_ack = htonl(state->remote_seq + data_size); #ifdef KEEPALIVE state->my_seq_sent += total_data_size; #else state->my_seq_sent += total_data_size + 1; //+1 for FIN tcp_header->tcp_flags = 0x11; state->fin_sent = 1; #endif char* new_data = (char*)tcp_header + sizeof(struct tcp_hdr); memcpy(new_data, g_http_part1, g_http_part1_size); new_data += g_http_part1_size; memcpy(new_data, content_length, content_length_size); new_data += content_length_size; memcpy(new_data, g_http_part2, g_http_part2_size); new_data += g_http_part2_size; memcpy(new_data, http->request_url, http->request_url_size); new_data += http->request_url_size; memcpy(new_data, g_http_part3, g_http_part3_size); memcpy(&http->last_message, (char*)tcp_header+sizeof(struct tcp_hdr), total_data_size); http->last_message_size = total_data_size; send_packet(tcp_header, packet); } else { ERROR("rte_pktmbuf_alloc, tcp data"); } http->state = HTTP_START; http->request_url_size = 0; } else if (*current == '\r') { http->state = HTTP_READ_R1; } else { http->state = HTTP_READ_URL; } break; } default: { ERROR("bad http state"); return; } } if (http->state == HTTP_BAD_STATE) { return; } --remaining_data; ++current; } }