stv0g/metalsvm
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

use BSD sockets instead of the RAW interface

Browse source
This commit is contained in:
Stefan Lankes 2011-09-14 07:50:12 -07:00
parent c73b69a2d1
commit 28429af30b
1 changed files with 267 additions and 355 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

622
kernel/netio.c
View file

@ -1,367 +1,279 @@
/*
* Redistribution and use in source and binary forms, with or without modification,
* are permitted provided that the following conditions are met:
/*
* Copyright 2011 Stefan Lankes, Chair for Operating Systems,
* RWTH Aachen University
*
* 1. Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* 2. Redistributions in binary form must reproduce the above copyright notice,
* this list of conditions and the following disclaimer in the documentation
* and/or other materials provided with the distribution.
* 3. The name of the author may not be used to endorse or promote products
* derived from this software without specific prior written permission.
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED
* WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
* MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT
* SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
* EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT
* OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING
* IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY
* OF SUCH DAMAGE.
* http://www.apache.org/licenses/LICENSE-2.0
*
* This file is part of the lwIP TCP/IP stack.
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
#include <metalsvm/stdlib.h>
#include <metalsvm/processor.h>
#if defined(CONFIG_LWIP)
#include <lwip/opt.h>
#if LWIP_TCP
#include <lwip/tcp.h>
/*
* This implements a netio server.
* The client sends a command word (4 bytes) then a data length word (4 bytes).
* If the command is "receive", the server is to consume "data length" bytes into
* a circular buffer until the first byte is non-zero, then it is to consume
* another command/data pair.
* If the command is "send", the server is to send "data length" bytes from a circular
* buffer with the first byte being zero, until "some time" and then send one final buffer with
* the first byte being non-zero. Then it is to consume another command/data pair.
* This file is part of MetalSVM.
*/
/* See http://www.nwlab.net/art/netio/netio.html to get the netio tool */
#include <metalsvm/stddef.h>
#include <metalsvm/stdio.h>
#include <metalsvm/time.h>
#include <metalsvm/tasks.h>
#include <metalsvm/syscall.h>
#include <metalsvm/errno.h>
/* implementation options */
#define NETIO_BUF_SIZE (4 * 1024)
#define NETIO_USE_STATIC_BUF 0
#ifdef CONFIG_LWIP
#include <lwip/sockets.h>
/* NetIO server state definition */
#define NETIO_STATE_WAIT_FOR_CMD 0
#define NETIO_STATE_RECV_DATA 1
#define NETIO_STATE_SEND_DATA 2
#define NETIO_STATE_SEND_DATA_LAST 3
#define NETIO_STATE_DONE 4
struct netio_state {
u32_t state;
u32_t cmd;
u32_t data_len;
u32_t cntr;
u8_t * buf_ptr;
u32_t buf_pos;
u32_t first_byte;
u64_t time_stamp;
};
/* NetIO command protocol definition */
#define NETIO_CMD_QUIT 0
#define NETIO_CMD_C2S 1
#define NETIO_CMD_S2C 2
#define NETIO_CMD_RES 3
static err_t netio_recv(void *arg, struct tcp_pcb *pcb, struct pbuf *p, err_t err);
static void
netio_close(void *arg, struct tcp_pcb *pcb)
typedef struct
{
err_t err;
uint32_t cmd;
uint32_t data;
} CONTROL;
struct netio_state *ns = arg;
ns->state = NETIO_STATE_DONE;
tcp_recv(pcb, NULL);
err = tcp_close(pcb);
if (err != ERR_OK) {
/* closing failed, try again later */
tcp_recv(pcb, netio_recv);
} else {
/* closing succeeded */
#if NETIO_USE_STATIC_BUF != 1
if(ns->buf_ptr != NULL){
mem_free(ns->buf_ptr);
}
#endif
tcp_arg(pcb, NULL);
tcp_poll(pcb, NULL, 0);
tcp_sent(pcb, NULL);
if (arg != NULL) {
mem_free(arg);
}
}
}
static err_t
netio_recv(void *arg, struct tcp_pcb *pcb, struct pbuf *p, err_t err)
{
struct netio_state *ns = arg;
u8_t * data_ptr;
u32_t data_cntr;
struct pbuf *q = p;
u16_t len;
if (err == ERR_OK && q != NULL) {
while (q != NULL) {
data_cntr = q->len;
data_ptr = q->payload;
while (data_cntr--) {
if (ns->state == NETIO_STATE_DONE){
netio_close(ns, pcb);
break;
} else if (ns->state == NETIO_STATE_WAIT_FOR_CMD) {
if (ns->cntr < 4) {
/* build up the CMD field */
ns->cmd <<= 8;
ns->cmd |= *data_ptr++;
ns->cntr++;
} else if (ns->cntr < 8) {
/* build up the DATA field */
ns->data_len <<= 8;
ns->data_len |= *data_ptr++;
ns->cntr++;
if (ns->cntr == 8) {
/* now we have full command and data words */
ns->cntr = 0;
ns->buf_pos = 0;
ns->buf_ptr[0] = 0;
if (ns->cmd == NETIO_CMD_C2S) {
ns->state = NETIO_STATE_RECV_DATA;
} else if (ns->cmd == NETIO_CMD_S2C) {
ns->state = NETIO_STATE_SEND_DATA;
/* start timer */
ns->time_stamp = rdtsc();
/* send first round of data */
len = tcp_sndbuf(pcb);
len = LWIP_MIN(len, ns->data_len - ns->cntr);
len = LWIP_MIN(len, NETIO_BUF_SIZE - ns->buf_pos);
//kprintf("data length: %d, len: %d\n", ns->data_len, len);
do {
err = tcp_write(pcb, ns->buf_ptr + ns->buf_pos, len, TCP_WRITE_FLAG_COPY);
if (err == ERR_MEM) {
len /= 2;
}
} while ((err == ERR_MEM) && (len > 1));
ns->buf_pos += len;
ns->cntr += len;
} else {
/* unrecognized command, punt */
ns->cntr = 0;
ns->buf_pos = 0;
ns->buf_ptr[0] = 0;
netio_close(ns, pcb);
break;
}
}
} else {
/* in trouble... shouldn't be in this state! */
kputs("netio is in trouble\n");
}
} else if (ns->state == NETIO_STATE_RECV_DATA) {
if(ns->cntr == 0){
/* save the first byte of this new round of data
* this will not match ns->buf_ptr[0] in the case that
* NETIO_BUF_SIZE is less than ns->data_len.
*/
ns->first_byte = *data_ptr;
}
ns->buf_ptr[ns->buf_pos++] = *data_ptr++;
ns->cntr++;
if (ns->buf_pos == NETIO_BUF_SIZE) {
/* circularize the buffer */
ns->buf_pos = 0;
}
if(ns->cntr == ns->data_len){
ns->cntr = 0;
if (ns->first_byte != 0) {
/* if this last round did not start with 0,
* go look for another command */
ns->state = NETIO_STATE_WAIT_FOR_CMD;
ns->data_len = 0;
ns->cmd = 0;
/* TODO LWIP_DEBUGF( print out some throughput calculation results... ); */
} else {
/* stay here and wait on more data */
}
}
} else if (ns->state == NETIO_STATE_SEND_DATA
|| ns->state == NETIO_STATE_SEND_DATA_LAST) {
/* I don't think this should happen... */
} else {
/* done / quit */
netio_close(ns, pcb);
break;
} /* end of ns->state condition */
} /* end of while data still in this pbuf */
q = q->next;
}
tcp_recved(pcb, p->tot_len);
pbuf_free(p);
} else {
/* error or closed by other side */
if (p != NULL) {
tcp_recved(pcb, p->tot_len);
pbuf_free(p);
}
/* close the connection */
netio_close(ns, pcb);
}
return ERR_OK;
}
static err_t
netio_sent(void *arg, struct tcp_pcb *pcb, u16_t len)
{
struct netio_state *ns = arg;
err_t err = ERR_OK;
if (ns->cntr >= ns->data_len && ns->state == NETIO_STATE_SEND_DATA) {
u64_t now = rdtsc();
u64_t time = (now - ns->time_stamp) / get_cpu_frequency();
/* done with this round of sending */
ns->buf_pos = 0;
ns->cntr = 0;
ns->buf_ptr[0] = 1;
ns->state = NETIO_STATE_SEND_DATA_LAST;
kprintf("consumed time: %llu usec (%llu ticks)\n", time, now - ns->time_stamp);
kprintf("throughput: %llu KByte/s\n", (ns->data_len * 1000000ULL) / (time*1024));
}
if(ns->state == NETIO_STATE_SEND_DATA_LAST || ns->state == NETIO_STATE_SEND_DATA){
len = tcp_sndbuf(pcb);
len = LWIP_MIN(len, ns->data_len - ns->cntr);
len = LWIP_MIN(len, NETIO_BUF_SIZE - ns->buf_pos);
if(ns->cntr < ns->data_len){
do {
err = tcp_write(pcb, ns->buf_ptr + ns->buf_pos, len, TCP_WRITE_FLAG_COPY);
if (err == ERR_MEM) {
len /= 2;
}
} while ((err == ERR_MEM) && (len > 1));
ns->buf_pos += len;
if(ns->buf_pos >= NETIO_BUF_SIZE){
ns->buf_pos = 0;
}
ns->cntr += len;
}
}
if(ns->cntr >= ns->data_len && ns->state == NETIO_STATE_SEND_DATA_LAST){
/* we have buffered up all our data to send this last round, go look for a command */
ns->state = NETIO_STATE_WAIT_FOR_CMD;
ns->cntr = 0;
ns->buf_ptr[0] = 0;
}
return ERR_OK;
}
static err_t
netio_poll(void *arg, struct tcp_pcb *pcb)
{
struct netio_state * ns = arg;
if(ns->state == NETIO_STATE_SEND_DATA) {
} else if(ns->state == NETIO_STATE_DONE){
netio_close(ns, pcb);
}
return ERR_OK;
}
#if NETIO_USE_STATIC_BUF == 1
static u8_t netio_buf[NETIO_BUF_SIZE];
#endif
static err_t
netio_accept(void *arg, struct tcp_pcb *pcb, err_t err)
{
struct netio_state * ns;
LWIP_UNUSED_ARG(err);
ns = mem_malloc(sizeof(struct netio_state));
if(ns == NULL){
return ERR_MEM;
}
ns->state = NETIO_STATE_WAIT_FOR_CMD;
ns->data_len = 0;
ns->cmd = 0;
ns->cntr = 0;
ns->buf_pos = 0;
#if NETIO_USE_STATIC_BUF == 1
ns->buf_ptr = netio_buf;
#else
ns->buf_ptr = mem_malloc(NETIO_BUF_SIZE);
if(ns->buf_ptr == NULL){
mem_free(ns);
return ERR_MEM;
}
#endif
memset(ns->buf_ptr, 0x00, NETIO_BUF_SIZE);
tcp_arg(pcb, ns);
tcp_sent(pcb, netio_sent);
tcp_recv(pcb, netio_recv);
tcp_poll(pcb, netio_poll, 4); /* every 2 seconds */
//tcp_nagle_disable(pcb);
return ERR_OK;
}
void netio_init(void)
{
struct tcp_pcb *pcb;
pcb = tcp_new();
tcp_bind(pcb, IP_ADDR_ANY, 18767);
pcb = tcp_listen(pcb);
tcp_accept(pcb, netio_accept);
}
#endif /* LWIP_TCP */
#define CMD_QUIT 0
#define CMD_C2S 1
#define CMD_S2C 2
#define CMD_RES 3
#define CTLSIZE sizeof(CONTROL)
#define DEFAULTPORT 0x494F /* "IO" */
#define TMAXSIZE 65536
//static int tSizes[] = {1024, 2048, 4096, 8192, 16384, 32767};
//static size_t ntSizes = sizeof(tSizes) / sizeof(int);
static int nPort = DEFAULTPORT;
static const int sobufsize = 131072;
static struct in_addr addr_local;
static int bTimeOver = 0;
static int send_data(int socket, void *buffer, size_t size, int flags)
{
int rc = send(socket, buffer, size, flags);
if (rc < 0)
{
kprintf("send failed: %d\n", rc);
return -1;
}
if (rc != size)
return 1;
return 0;
}
static int recv_data(int socket, void *buffer, size_t size, int flags)
{
size_t rc = recv(socket, buffer, size, flags);
if (rc < 0) {
kprintf("recv failed: %d\n", rc);
return -1;
}
if (rc != size)
return 1;
return 0;
}
static char *InitBuffer(size_t nSize)
{
char *cBuffer = kmalloc(nSize);
memset(cBuffer, 0xFF, nSize);
cBuffer[0] = 0;
return cBuffer;
}
static char *PacketSize(int nSize)
{
static char szBuffer[64];
if ((nSize % 1024) == 0 || (nSize % 1024) == 1023)
ksprintf(szBuffer, "%2dk", (nSize + 512) / 1024);
else
ksprintf(szBuffer, "%d", nSize);
return szBuffer;
}
static int TCPServer(void* arg)
{
char *cBuffer;
CONTROL ctl;
uint64_t nData;
struct sockaddr_in sa_server, sa_client;
int server, client;
socklen_t length;
struct timeval tv;
fd_set fds;
int rc;
int nByte;
int err;
uint64_t start, end;
uint32_t freq = get_cpu_frequency(); /* in MHz */
if ((cBuffer = InitBuffer(TMAXSIZE)) == NULL) {
kprintf("Netio: Not enough memory\n");
return -EINVAL;
}
if ((server = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
kprintf("socket failed: %d\n", server);
kfree(cBuffer, TMAXSIZE);
return -1;
}
setsockopt(server, SOL_SOCKET, SO_RCVBUF, (char *) &sobufsize, sizeof(sobufsize));
setsockopt(server, SOL_SOCKET, SO_SNDBUF, (char *) &sobufsize, sizeof(sobufsize));
sa_server.sin_family = AF_INET;
sa_server.sin_port = htons(nPort);
sa_server.sin_addr = addr_local;
if ((err = bind(server, (struct sockaddr *) &sa_server, sizeof(sa_server))) < 0)
{
kprintf("bind failed: %d\n", err);
close(server);
kfree(cBuffer, TMAXSIZE);
return -1;
}
if ((err = listen(server, 2)) != 0)
{
kprintf("listen failed: %d\n", err);
close(server);
kfree(cBuffer, TMAXSIZE);
return -1;
}
for (;;)
{
kprintf("TCP server listening.\n");
FD_ZERO(&fds);
FD_SET(server, &fds);
tv.tv_sec = 3600;
tv.tv_usec = 0;
if ((rc = select(FD_SETSIZE, &fds, 0, 0, &tv)) < 0)
{
kprintf("select failed: %d\n", rc);
break;
}
if (rc == 0 || FD_ISSET(server, &fds) == 0)
continue;
length = sizeof(sa_client);
if ((client = accept(server, (struct sockaddr *) &sa_client, &length)) == -1)
continue;
setsockopt(client, SOL_SOCKET, SO_RCVBUF, (char *) &sobufsize, sizeof(sobufsize));
setsockopt(client, SOL_SOCKET, SO_SNDBUF, (char *) &sobufsize, sizeof(sobufsize));
kprintf("TCP connection established ... ");
for (;;)
{
if (recv_data(client, (void *) &ctl, CTLSIZE, 0))
break;
ctl.cmd = ntohl(ctl.cmd);
ctl.data = ntohl(ctl.data);
if (ctl.cmd == CMD_C2S)
{
start = rdtsc();
kprintf("\nReceiving from client, packet size %s ... ", PacketSize(ctl.data));
cBuffer[0] = 0;
nData = 0;
do {
for (nByte = 0; nByte < ctl.data; )
{
rc = recv(client, cBuffer + nByte, ctl.data - nByte, 0);
if (rc < 0)
{
kprintf("recv failed: %d\n", rc);
break;
}
if (rc > 0)
nByte += rc;
}
nData += ctl.data;
} while (cBuffer[0] == 0 && rc > 0);
end = rdtsc();
kprintf("Time to receive %llu bytes: %llu nsec (ticks %llu)\n", nData, ((end-start)*1000ULL)/freq, end-start);
} else if (ctl.cmd == CMD_S2C) {
start = rdtsc();
kprintf("\nSending to client, packet size %s ... ", PacketSize(ctl.data));
cBuffer[0] = 0;
nData = 0;
do
{
//GenerateRandomData(cBuffer, ctl.data);
for (nByte = 0; nByte < ctl.data; )
{
rc = send(client, cBuffer + nByte, ctl.data - nByte, 0);
if (rc < 0)
{
kprintf("send failed: %d\n", rc);
break;
}
if (rc > 0)
nByte += rc;
}
kputs("B");
nData += ctl.data;
} while(((rdtsc()-start)/(uint64_t)freq) < 6000000ULL /*= 6s */);
kputs("AAAA");
cBuffer[0] = 1;
if (send_data(client, cBuffer, ctl.data, 0))
break;
end = rdtsc();
kprintf("Time to send %llu bytes: %llu nsec (ticks %llu)\n", nData, ((end-start)*1000ULL)/freq, end-start);
} else /* quit */
break;
}
kprintf("\nDone.\n");
close(client);
if (rc < 0)
break;
}
close(server);
kfree(cBuffer, TMAXSIZE);
return 0;
}
int netio_init(void)
{
addr_local.s_addr = INADDR_ANY;
return create_kernel_task(NULL, TCPServer, NULL, NORMAL_PRIO);
}
#endif