From: Peter Dinda Date: Wed, 26 Dec 2012 17:10:36 +0000 (-0600) Subject: Implemntation of newer keyed stream interface on linux, includes new "textfile" ... X-Git-Url: http://v3vee.org/palacios/gitweb/gitweb.cgi?a=commitdiff_plain;h=840ee2db0091766efea20481af7c3ac6f3a77e93;p=palacios.releases.git Implemntation of newer keyed stream interface on linux, includes new "textfile" (ini file) output capability --- diff --git a/linux_module/iface-keyed-stream-user.h b/linux_module/iface-keyed-stream-user.h index 84fd844..85ab0e6 100644 --- a/linux_module/iface-keyed-stream-user.h +++ b/linux_module/iface-keyed-stream-user.h @@ -35,6 +35,10 @@ struct palacios_user_keyed_stream_url { // This structure is used for both requests (kernel->user) // and responses (user->kernel) // +// for a readkey request, the buf contains the tag +// for a readkey response, the buf contains the data +// for a writekey request, the buf contains the data + key +// for a writekey request, the buf contains nothing struct palacios_user_keyed_stream_op { uint64_t len; // total structure length (all) @@ -48,13 +52,17 @@ struct palacios_user_keyed_stream_op { #define PALACIOS_KSTREAM_READ_KEY 6 sint64_t xfer; // total bytes read or written (request/response) + // void *user_key; // user tag for an open key (response) uint64_t buf_len; // buffer len - char buf[0]; // expanded as needed (key or valye) + uint64_t data_off; // offset of data within the buffer + // 0..buffer_len-1 is tag + // rest is data + char buf[0]; // expanded as needed (key or value) - // The buffer contains the key or the value + // The buffer contains the key or the value }; diff --git a/linux_module/iface-keyed-stream.c b/linux_module/iface-keyed-stream.c index 41da2fb..75978ab 100644 --- a/linux_module/iface-keyed-stream.c +++ b/linux_module/iface-keyed-stream.c @@ -6,6 +6,7 @@ * (c) Peter Dinda, 2011 (interface, mem + file implementations + recooked user impl) * (c) Clint Sbisa, 2011 (initial user space implementation on which this is based) * (c) Diana Palsetia & Steve Rangel, 2012 (network based implementation) + * (c) Peter Dinda, 2012 (updated interface, textfile) */ #include @@ -54,21 +55,57 @@ "mem:" Streams are stored in a hash table The values for this hash table are hash tables associated with - each stream. + each stream. A key maps to an expanding memory buffer. + Data is stored in a buffer like: + + [boundarytag][taglen][tag][datalen][data] + [boundarytag][taglen][tag][datalen][data] + ... "file:" Streams are stored in files. Each high-level open corresponds to a directory, while a key corresponds to - a distinct file in that directory. + a distinct file in that directory. Data is stored in a file + like: + + [boundarytag][taglen][tag][datalen][data] + [boundarytag][taglen][tag][datalen][data] + ... + + "textfile:" Same as file, but data is stored in text format, like a + windows style .ini file. A key maps to a file, and data is stored + in a file like: + + [key] + tag=data_in_hex + tag=data_in_hex + + This format makes it possible to concentenate the files to + produce a single "ini" file with "sections". + "net:" Streams are carried over the network. Each high level open corresponds to a TCP connection, while each key corresponds to a context on the stream. "net:a::" => Bind to : and accept a connection "net:c::" => Connect to : + "c" (client) + open_stream: connect + "a" (server) + open_stream: accept + "c" or "a": + open_key: send [keylen-lastbyte-high-bit][key] (writer) + or recv (same format as above) (reader) + close_key: send [keylen-lastbyte-high-bit][key] (writer) + or recv (same format as above) (reader) + write_key: send [boundarytag][taglen][tag][datalen][data] + read_key: recv (same format as above) + close_stream: close socket "user:" Stream requests are bounced to user space to be handled there. A rendezvous approach similar to the host device userland support is used + + All keyed streams store the tags. */ @@ -77,6 +114,7 @@ #define STREAM_FILE 2 #define STREAM_USER 3 #define STREAM_NETWORK 4 +#define STREAM_TEXTFILE 5 /* All keyed streams and streams indicate their implementation type within the first field @@ -89,7 +127,10 @@ struct generic_stream { int stype; }; - +/* + boundary tags are used for some othe raw formats. +*/ +static uint32_t BOUNDARY_TAG=0xabcd0123; /**************************************************************************************** @@ -200,6 +241,8 @@ static uint32_t write_mem_stream(struct mem_stream *m, } + + static uint32_t read_mem_stream(struct mem_stream *m, void *data, uint32_t len) @@ -425,62 +468,173 @@ static void close_key_mem(v3_keyed_stream_t stream, static sint64_t write_key_mem(v3_keyed_stream_t stream, v3_keyed_stream_key_t key, + void *tag, + sint64_t taglen, void *buf, sint64_t len) { - struct mem_keyed_stream *mks = (struct mem_keyed_stream *) stream; - struct mem_stream *m = (struct mem_stream *) key; - uint32_t mylen; - uint32_t writelen; - - if (mks->ot!=V3_KS_WR_ONLY) { - return -1; - } + struct mem_keyed_stream *mks = (struct mem_keyed_stream *) stream; + struct mem_stream *m = (struct mem_stream *) key; + uint32_t mylen; + uint32_t writelen; + + if (mks->ot!=V3_KS_WR_ONLY) { + return -1; + } + + if (taglen<0 || len<0) { + ERROR("Negative taglen or data len\n"); + return -1; + } + + if (taglen>0xffffffffULL || len>0xffffffffULL) { + ERROR("taglen or data len is too large\n"); + return -1; + } + + writelen=write_mem_stream(m,&BOUNDARY_TAG,sizeof(BOUNDARY_TAG)); + + if (writelen!=sizeof(BOUNDARY_TAG)) { + ERROR("failed to write all data for boundary tag\n"); + return -1; + } + + writelen=write_mem_stream(m,&taglen,sizeof(taglen)); + + if (writelen!=sizeof(taglen)) { + ERROR("failed to write taglen\n"); + return -1; + } + + mylen = (uint32_t) taglen; + + writelen=write_mem_stream(m,tag,mylen); + + if (writelen!=mylen) { + ERROR("failed to write all data for tag\n"); + return -1; + } - if (len<0) { - return -1; - } - - mylen = (uint32_t) len; + writelen=write_mem_stream(m,&len,sizeof(len)); + + if (writelen!=sizeof(len)) { + ERROR("failed to write datalen\n"); + return -1; + } + + mylen = (uint32_t) len; - writelen=write_mem_stream(m,buf,mylen); + writelen=write_mem_stream(m,buf,mylen); - if (writelen!=mylen) { - ERROR("failed to write all data for key\n"); - return -1; - } else { - return (sint64_t)writelen; - } + if (writelen!=mylen) { + ERROR("failed to write all data for key\n"); + return -1; + } else { + return (sint64_t)writelen; + } } static sint64_t read_key_mem(v3_keyed_stream_t stream, v3_keyed_stream_key_t key, + void *tag, + sint64_t taglen, void *buf, sint64_t len) { - struct mem_keyed_stream *mks = (struct mem_keyed_stream *) stream; - struct mem_stream *m = (struct mem_stream *) key; - uint32_t mylen; - uint32_t readlen; - - if (mks->ot!=V3_KS_RD_ONLY) { - return -1; - } + struct mem_keyed_stream *mks = (struct mem_keyed_stream *) stream; + struct mem_stream *m = (struct mem_stream *) key; + uint32_t mylen; + uint32_t readlen; + void *temptag; + uint32_t tempbt; + sint64_t templen; + + + if (mks->ot!=V3_KS_RD_ONLY) { + return -1; + } + + if (len<0 || taglen<0) { + ERROR("taglen or data len is negative\n"); + return -1; + } - if (len<0) { - return -1; - } + if (len>0xffffffffULL || taglen>0xffffffffULL) { + ERROR("taglen or data len is too large\n"); + return -1; + } + + readlen=read_mem_stream(m,&tempbt,sizeof(tempbt)); + + if (readlen!=sizeof(tempbt)) { + ERROR("failed to read all data for boundary tag\n"); + return -1; + } + + if (tempbt!=BOUNDARY_TAG) { + ERROR("boundary tag not found (read 0x%x)\n",tempbt); + return -1; + } + + readlen=read_mem_stream(m,&templen,sizeof(templen)); + + if (readlen!=sizeof(templen)) { + ERROR("failed to read all data for taglen\n"); + return -1; + } + + if (templen!=taglen) { + ERROR("tag size mismatch (requested=%lld, actual=%lld)\n",taglen,templen); + return -1; + } + + temptag = palacios_alloc(taglen); - mylen = (uint32_t) len; + if (!temptag) { + ERROR("cannot allocate temporary tag\n"); + return -1; + } + + mylen = (uint32_t) len; - readlen=read_mem_stream(m,buf,mylen); + readlen=read_mem_stream(m,temptag,mylen); - if (readlen!=mylen) { - ERROR("failed to read all data for key\n"); - return -1; - } else { - return (sint64_t)readlen; - } + if (readlen!=mylen) { + ERROR("failed to read all data for tag\n"); + palacios_free(temptag); + return -1; + } + + if (memcmp(tag,temptag,taglen)) { + ERROR("tag mismatch\n"); + palacios_free(temptag); + return -1; + } + + palacios_free(temptag); + + readlen=read_mem_stream(m,&templen,sizeof(templen)); + + if (readlen!=sizeof(templen)) { + ERROR("failed to read all data for data len\n"); + return -1; + } + + if (templen!=len) { + ERROR("data size mismatch (requested=%lld, actual=%lld)\n",len,templen); + return -1; + } + + mylen = (uint32_t) len; + + readlen=read_mem_stream(m,buf,mylen); + + if (readlen!=mylen) { + ERROR("failed to read all data for key\n"); + return -1; + } else { + return (sint64_t)readlen; + } } @@ -682,85 +836,797 @@ static void close_key_file(v3_keyed_stream_t stream, palacios_free(fs); } + +static sint64_t write_file(struct file_stream *fs, void *buf, sint64_t len) +{ + ssize_t done, left, total; + mm_segment_t old_fs; + + total=len; + left=len; + + while (left>0) { + old_fs = get_fs(); + set_fs(get_ds()); + done = fs->f->f_op->write(fs->f, buf+(total-left), left, &(fs->f->f_pos)); + set_fs(old_fs); + if (done<=0) { + return -1; + } else { + left -= done; + } + } + + return len; +} + static sint64_t write_key_file(v3_keyed_stream_t stream, v3_keyed_stream_key_t key, + void *tag, + sint64_t taglen, void *buf, sint64_t len) { - struct file_keyed_stream *fks = (struct file_keyed_stream *) stream; - struct file_stream *fs = (struct file_stream *) key; - mm_segment_t old_fs; + struct file_keyed_stream *fks = (struct file_keyed_stream *) stream; + struct file_stream *fs = (struct file_stream *) key; + sint64_t writelen; + + if (fks->ot!=V3_KS_WR_ONLY) { + return -1; + } + + if (taglen<0 || len<0) { + ERROR("Negative taglen or data len\n"); + return -1; + } + + writelen=write_file(fs,&BOUNDARY_TAG,sizeof(BOUNDARY_TAG)); + + if (writelen!=sizeof(BOUNDARY_TAG)) { + ERROR("failed to write all data for boundary tag\n"); + return -1; + } + + writelen=write_file(fs,&taglen,sizeof(taglen)); + + if (writelen!=sizeof(taglen)) { + ERROR("failed to write taglen\n"); + return -1; + } + + if (write_file(fs,tag,taglen)!=taglen) { + ERROR("failed to write tag\n"); + return -1; + } + + writelen=write_file(fs,&len,sizeof(len)); + + if (writelen!=sizeof(len)) { + ERROR("failed to write data len\n"); + return -1; + } + + return write_file(fs,buf,len); +} + +static sint64_t read_file(struct file_stream *fs, void *buf, sint64_t len) +{ ssize_t done, left, total; - - if (fks->ot!=V3_KS_WR_ONLY) { - return -1; - } - - if (len<0) { - return -1; - } + mm_segment_t old_fs; total=len; left=len; - old_fs = get_fs(); - set_fs(get_ds()); while (left>0) { - done = fs->f->f_op->write(fs->f, buf+(total-left), left, &(fs->f->f_pos)); + old_fs = get_fs(); + set_fs(get_ds()); + done = fs->f->f_op->read(fs->f, buf+(total-left), left, &(fs->f->f_pos)); + set_fs(old_fs); if (done<=0) { return -1; } else { left -= done; } } - set_fs(old_fs); return len; } - static sint64_t read_key_file(v3_keyed_stream_t stream, v3_keyed_stream_key_t key, + void *tag, + sint64_t taglen, void *buf, sint64_t len) { - struct file_keyed_stream *fks = (struct file_keyed_stream *) stream; - struct file_stream *fs = (struct file_stream *) key; - mm_segment_t old_fs; - ssize_t done, left, total; - - if (fks->ot!=V3_KS_RD_ONLY) { + struct file_keyed_stream *fks = (struct file_keyed_stream *) stream; + struct file_stream *fs = (struct file_stream *) key; + void *temptag; + uint32_t tempbt; + sint64_t templen; + sint64_t readlen; + + if (fks->ot!=V3_KS_RD_ONLY) { + return -1; + } + + if (len<0 || taglen<0) { + ERROR("taglen or data len is negative\n"); + return -1; + } + + readlen=read_file(fs,&tempbt,sizeof(tempbt)); + + if (readlen!=sizeof(tempbt)) { + ERROR("failed to read all data for boundary tag\n"); + return -1; + } + + if (tempbt!=BOUNDARY_TAG) { + ERROR("boundary tag not found (read 0x%x)\n",tempbt); + return -1; + } + + readlen=read_file(fs,&templen,sizeof(templen)); + + if (readlen!=sizeof(templen)) { + ERROR("failed to read all data for tag len\n"); + return -1; + } + + if (templen!=taglen) { + ERROR("tag size mismatch (requested=%lld, actual=%lld)\n",taglen,templen); + return -1; + } + + temptag=palacios_alloc(taglen); + + if (!temptag) { + ERROR("Cannot allocate temptag\n"); + return -1; + } + + if (read_file(fs,temptag,taglen)!=taglen) { + ERROR("Cannot read tag\n"); + palacios_free(temptag); + return -1; + } + + if (memcmp(temptag,tag,taglen)) { + ERROR("Tag mismatch\n"); + palacios_free(temptag); + return -1; + } + + palacios_free(temptag); + + readlen=read_file(fs,&templen,sizeof(templen)); + + if (readlen!=sizeof(templen)) { + ERROR("failed to read all data for data len\n"); + return -1; + } + + if (templen!=len) { + ERROR("datasize mismatch (requested=%lld, actual=%lld)\n",len,templen); + return -1; + } + + return read_file(fs,buf,len); + +} + + +/*************************************************************************************************** + Textfile-based implementation ("textfile:") + + Note that this implementation uses the internal structure and functions of the + "file:" implementation +*************************************************************************************************/ + +// optimize the reading and decoding of hex data +// this weakens the parser, so that: +// tag =0A0B0D +// will work, but +// tag = 0A 0B 0D +// will not. Note the leading whitespace +#define TEXTFILE_OPT_HEX 0 + +// +// The number of bytes handled at a time by the hex putter and getter +// +#define MAX_HEX_SEQ 64 + +/* + A text file keyed stream is a file_keyed_stream, + only with a different stype +*/ + +#define PAUSE() +//#define PAUSE() ssleep(5) + + +typedef struct file_keyed_stream textfile_keyed_stream; + +typedef struct file_stream textfile_stream; + + +static v3_keyed_stream_t open_stream_textfile(char *url, + v3_keyed_stream_open_t ot) +{ + textfile_keyed_stream *me; + + if (strncasecmp(url,"textfile:",9)) { + WARNING("illegitimate attempt to open textfile stream \"%s\"\n",url); + return 0; + } + + me = (textfile_keyed_stream *) open_stream_file(url+4, ot); + + if (!me) { + ERROR("could not create underlying file stream\n"); + return 0; + } + + me->stype=STREAM_TEXTFILE; + + return me; +} + + + +static void close_stream_textfile(v3_keyed_stream_t stream) +{ + textfile_keyed_stream *me = stream; + + me->stype=STREAM_FILE; + + close_stream_file(me); + +} + +static void preallocate_hint_key_textfile(v3_keyed_stream_t stream, + char *key, + uint64_t size) +{ + textfile_keyed_stream *me = stream; + + me->stype=STREAM_FILE; + + preallocate_hint_key_file(me,key,size); + + me->stype=STREAM_TEXTFILE; + +} + + +static inline int isoneof(char c, char *sl, int m) +{ + int i; + + for (i=0;i=0 && c<=9) { + return '0'+c; + } else if (c>=0xa && c<=0xf) { + return 'a'+(c-0xa); + } else { + return -1; + } +} + +static int hexify_byte(char *c, char b) +{ + char n; + n = hexify_nybble( (b >> 4) & 0xf); + if (n==-1) { + return -1; + } + c[0] = n; + n = hexify_nybble( b & 0xf); + if (n==-1) { + return -1; + } + c[1] = n; + return 0; +} + + +static char dehexify_nybble(char c) +{ + if (c>='0' && c<='9') { + return c-'0'; + } else if (c>='a' && c<='f') { + return 0xa + (c-'a'); + } else if (c>='A' && c<='F') { + return 0xa + (c-'A'); + } else { + return -1; + } +} + +static int dehexify_byte(char *c, char *b) +{ + char n; + n = dehexify_nybble(c[0]); + if (n==-1) { + return -1; + } + *b = n << 4; + n = dehexify_nybble(c[1]); + if (n==-1) { + return -1; + } + *b |= n; + return 0; +} + + +#if TEXTFILE_OPT_HEX + + +// Here the sl array, of length m is the number +static int get_hexbytes_as_data(textfile_stream *s, char *buf, int n) +{ + char rbuf[MAX_HEX_SEQ*2]; + int left = n; + int off = 0; + int cur = 0; + int i; + + while (left>0) { + cur = left > MAX_HEX_SEQ ? MAX_HEX_SEQ : left; + if (read_file(s,rbuf,cur*2)!=cur*2) { + ERROR("Cannot read data in getting hexbytes as data\n"); + return -1; + } + + for (i=0;i0) { - done = fs->f->f_op->read(fs->f, buf+(total-left), left, &(fs->f->f_pos)); - if (done<=0) { - return -1; - } else { - left -= done; - } +static int put_data_as_hexbytes(textfile_stream *s, char *buf, int n) +{ + char rbuf[MAX_HEX_SEQ*2]; + int left = n; + int off = 0; + int cur = 0; + int i; + + while (left>0) { + cur = left > MAX_HEX_SEQ ? MAX_HEX_SEQ : left; + for (i=0;istype=STREAM_FILE; + + ms = open_key_file(mks,key); + + if (!ms) { + ERROR("cannot open underlying file keyed stream for key %s\n",key); + mks->stype=STREAM_TEXTFILE; + return 0; + } + + if (mks->ot==V3_KS_WR_ONLY) { + + // Now we write the section header + + ms->stype=STREAM_FILE; + + if (put_string(ms,"[")) { + close_key_file(mks,ms); + mks->stype=STREAM_TEXTFILE; + return 0; + } + + if (put_string(ms,key)) { + close_key_file(mks,ms); + mks->stype=STREAM_TEXTFILE; + return 0; + } + + if (put_string(ms,"]\n")) { + close_key_file(mks,ms); + mks->stype=STREAM_TEXTFILE; + return 0; + } + + + mks->stype=STREAM_TEXTFILE; + ms->stype=STREAM_TEXTFILE; + + return ms; + + } else if (mks->ot == V3_KS_RD_ONLY) { + // Now we readthe section header + int keylen=strlen(key); + char *tempkey = palacios_alloc(keylen+3); + + ms->stype=STREAM_FILE; + + if (!tempkey) { + ERROR("Allocation failed in opening key\n"); + close_key_file(mks,ms); + mks->stype=STREAM_FILE; + return 0; + } + + + if (token_scan(ms,tempkey,keylen+3," \t\r\n",4)) { + ERROR("Cannot scan for token (key search)\n"); + close_key_file(mks,ms); + mks->stype=STREAM_TEXTFILE; + palacios_free(tempkey); + return 0; + } + + tempkey[keylen+2] = 0; + + // Should now have [key]0 + + if (tempkey[0]!='[' || + tempkey[keylen+1]!=']' || + memcmp(key,tempkey+1,keylen)) { + ERROR("key mismatch: target key=%s, found %s\n",key,tempkey); + palacios_free(tempkey); + close_key_file(mks,ms); + mks->stype=STREAM_TEXTFILE; + return 0; + } + + // key match done, success + + mks->stype=STREAM_TEXTFILE; + ms->stype=STREAM_TEXTFILE; + + palacios_free(tempkey); + + return ms; + + } else { + ERROR("Unknown open type in open_key_textfile\n"); + ms->stype=STREAM_FILE; + close_key_file(mks,ms); + return 0; + } + +} + + + +static void close_key_textfile(v3_keyed_stream_t stream, + v3_keyed_stream_key_t key) +{ + textfile_keyed_stream *mks = stream; + textfile_stream *ms=key; + + mks->stype=STREAM_FILE; + ms->stype=STREAM_FILE; + + close_key_file(mks,ms); + + mks->stype=STREAM_TEXTFILE; + +} + + +static sint64_t read_key_textfile(v3_keyed_stream_t stream, + v3_keyed_stream_key_t key, + void *tag, + sint64_t taglen, + void *buf, + sint64_t len) +{ + textfile_stream *ms = (textfile_stream *) key; + char tags[32]; + char *temptag; + + + + memcpy(tags,tag,taglen<31 ? taglen : 31); + tags[taglen<32? taglen : 31 ]=0; + + temptag=palacios_alloc(taglen+1); + if (!temptag) { + ERROR("Unable to allocate temptag in textfile read key\n"); + return -1; + } + + ms->stype=STREAM_FILE; + + if (token_scan(ms,temptag,taglen+1," \t\r\n=",5)) { + ERROR("Cannot scan for token (tag search)\n"); + ms->stype=STREAM_TEXTFILE; + palacios_free(temptag); + return -1; + } + + if (memcmp(tag,temptag,taglen)) { + ERROR("Tag mismatch in reading tag from textfile: desired tag=%s, actual tag=%s\n",tags,temptag); + ms->stype=STREAM_TEXTFILE; + palacios_free(temptag); + return -1; + } + + // tag matches, let's go and find our = + palacios_free(temptag); + + if (search_for(ms,'=')) { + ERROR("Unable to find = sign in tag data parse (tag=%s)\n", tags); + ms->stype=STREAM_TEXTFILE; + return -1; + } + + +#if TEXTFILE_OPT_HEX + if (get_hexbytes_as_data(ms,buf,len)) { + ERROR("Cannot read data in hex format (opt path) in textfile for tag %s\n",tags); + ms->stype=STREAM_TEXTFILE; + return -1; + } +#else + if (get_hexbytes_as_data_skip(ms,buf,len," \t\r\n",4)) { + ERROR("Cannot read data in hex format (unopt path) in textfile for tag %s\n",tags); + ms->stype=STREAM_TEXTFILE; + return -1; + } +#endif + + ms->stype=STREAM_TEXTFILE; + + return len; +} + +static sint64_t write_key_textfile(v3_keyed_stream_t stream, + v3_keyed_stream_key_t key, + void *tag, + sint64_t taglen, + void *buf, + sint64_t len) +{ + textfile_stream *ms = (textfile_stream *) key; + char tags[32]; + + + + memcpy(tags,tag,taglen<31 ? taglen : 31); + tags[taglen<32? taglen : 31 ]=0; + + /* if (taglen>100000 || len>100000) { + ERROR("Too big\n"); + return -1; + } + */ + + ms->stype=STREAM_FILE; + + if (put_string_n(ms,tag,taglen)) { + ERROR("Cannot write tag %s in textfile\n",tags); + ms->stype=STREAM_TEXTFILE; + return -1; + } + + if (put_string(ms,"=")) { + ERROR("Cannot write = in textfile for tag %s\n",tags); + ms->stype=STREAM_TEXTFILE; + return -1; + } + + if (put_data_as_hexbytes(ms,buf,len)) { + ERROR("Cannot write data in hex format in textfile for tag %s\n",tags); + ms->stype=STREAM_TEXTFILE; + return -1; + } + + if (put_string(ms,"\n")) { + ERROR("Cannot write trailing lf in textfile for tag %s\n",tags); + ms->stype=STREAM_TEXTFILE; + return -1; + } + + ms->stype=STREAM_TEXTFILE; + + return len; +} + + /*************************************************************************************************** User implementation ("user:") @@ -1268,11 +2134,13 @@ static void close_key_user(v3_keyed_stream_t stream, v3_keyed_stream_key_t key) static sint64_t read_key_user(v3_keyed_stream_t stream, v3_keyed_stream_key_t key, + void *tag, + sint64_t taglen, void *buf, sint64_t rlen) { struct user_keyed_stream *s = (struct user_keyed_stream *) stream; - uint64_t len = 0 ; + uint64_t len = taglen ; sint64_t xfer; unsigned long flags; @@ -1283,6 +2151,7 @@ static sint64_t read_key_user(v3_keyed_stream_t stream, v3_keyed_stream_key_t ke ERROR("attempt to read key from stream that is not in read state on %s\n",s->url); } + if (resize_op(&(s->op),len)) { spin_unlock_irqrestore(&(s->lock),flags); ERROR("cannot resize op in reading key 0x%p on user keyed stream %s\n",key,s->url); @@ -1293,6 +2162,9 @@ static sint64_t read_key_user(v3_keyed_stream_t stream, v3_keyed_stream_key_t ke s->op->buf_len = len ; s->op->xfer = rlen; s->op->user_key = key; + s->op->data_off = taglen; + + memcpy(s->op->buf,tag,taglen); // enter with it locked if (do_request_to_response(s,&flags)) { @@ -1304,6 +2176,7 @@ static sint64_t read_key_user(v3_keyed_stream_t stream, v3_keyed_stream_key_t ke if (s->op->xfer>0) { + // data_off must be zero memcpy(buf,s->op->buf,s->op->xfer); } @@ -1316,15 +2189,17 @@ static sint64_t read_key_user(v3_keyed_stream_t stream, v3_keyed_stream_key_t ke static sint64_t write_key_user(v3_keyed_stream_t stream, v3_keyed_stream_key_t key, + void *tag, + sint64_t taglen, void *buf, sint64_t wlen) { struct user_keyed_stream *s = (struct user_keyed_stream *) stream; - struct palacios_user_keyed_stream_op *op; - uint64_t len = wlen ; + uint64_t len = taglen + wlen ; sint64_t xfer; unsigned long flags; - + + spin_lock_irqsave(&(s->lock), flags); if (s->otype != V3_KS_WR_ONLY) { @@ -1338,14 +2213,14 @@ static sint64_t write_key_user(v3_keyed_stream_t stream, v3_keyed_stream_key_t k return -1; } - op = s->op; - s->op->type = PALACIOS_KSTREAM_WRITE_KEY; s->op->buf_len = len; s->op->xfer = wlen; s->op->user_key = key; + s->op->data_off = taglen; - memcpy(s->op->buf,buf,wlen); + memcpy(s->op->buf,tag,taglen); + memcpy(s->op->buf+taglen,buf,wlen); // enter with it locked if (do_request_to_response(s,&flags)) { @@ -1355,6 +2230,8 @@ static sint64_t write_key_user(v3_keyed_stream_t stream, v3_keyed_stream_key_t k } // return with it locked + // no data comes back, xfer should be size of data write (not tag) + xfer=s->op->xfer; spin_unlock_irqrestore(&(s->lock),flags); @@ -1856,7 +2733,10 @@ static void close_key_net(v3_keyed_stream_t stream, v3_keyed_stream_key_t input_ } } -static sint64_t write_key_net(v3_keyed_stream_t stream, v3_keyed_stream_key_t key, void *buf, sint64_t len) +static sint64_t write_key_net(v3_keyed_stream_t stream, v3_keyed_stream_key_t key, + void *tag, + sint64_t taglen, + void *buf, sint64_t len) { struct net_keyed_stream * nks = (struct net_keyed_stream *)stream; @@ -1865,10 +2745,20 @@ static sint64_t write_key_net(v3_keyed_stream_t stream, v3_keyed_stream_key_t ke return -1; } + if (!tag) { + ERROR("Tag is NULL in write_key_net\n"); + return -1; + } + if (len<0) { ERROR("len is negative in write_key_net\n"); return -1; } + + if (taglen<0) { + ERROR("taglen is negative in write_key_net\n"); + return -1; + } if (!key){ ERROR("write_key: key is NULL in write_key_net\n"); @@ -1882,6 +2772,18 @@ static sint64_t write_key_net(v3_keyed_stream_t stream, v3_keyed_stream_key_t ke } if (nks->ot==V3_KS_WR_ONLY) { + if (send_msg(nks->ns,(char*)(&BOUNDARY_TAG),sizeof(BOUNDARY_TAG))!=sizeof(BOUNDARY_TAG)) { + ERROR("Could not send boundary tag in write_key_net\n"); + return -1; + } + if (send_msg(nks->ns,(char*)(&taglen),sizeof(taglen))!=sizeof(taglen)) { + ERROR("Could not send tag length in write_key_net\n"); + return -1; + } + if (send_msg(nks->ns,tag,taglen)!=len) { + ERROR("Could not send tag in write_key_net\n"); + return -1; + } if (send_msg(nks->ns,(char*)(&len),sizeof(len))!=sizeof(len)) { ERROR("Could not send data length in write_key_net\n"); return -1; @@ -1899,19 +2801,33 @@ static sint64_t write_key_net(v3_keyed_stream_t stream, v3_keyed_stream_key_t ke } -static sint64_t read_key_net(v3_keyed_stream_t stream, v3_keyed_stream_key_t key, void *buf, sint64_t len) +static sint64_t read_key_net(v3_keyed_stream_t stream, v3_keyed_stream_key_t key, + void *tag, + sint64_t taglen, + void *buf, sint64_t len) { struct net_keyed_stream * nks = (struct net_keyed_stream *)stream; + void *temptag; if (!buf) { ERROR("Buf is NULL in read_key_net\n"); return -1; } + + if (!tag) { + ERROR("Tag is NULL in read_key_net\n"); + return -1; + } if(len<0) { ERROR("len is negative in read_key_net\n"); return -1; } + + if(taglen<0) { + ERROR("taglen is negative in read_key_net\n"); + return -1; + } if (!key) { ERROR("read_key: key is NULL in read_key_net\n"); @@ -1920,23 +2836,69 @@ static sint64_t read_key_net(v3_keyed_stream_t stream, v3_keyed_stream_key_t key if (nks->ot==V3_KS_RD_ONLY) { + + sint64_t slen; + uint32_t tempbt; + + if (recv_msg(nks->ns,(char*)(&tempbt),sizeof(tempbt))!=sizeof(tempbt)) { + ERROR("Cannot receive boundary tag in read_key_net\n"); + return -1; + } - sint64_t slen; + if (tempbt!=BOUNDARY_TAG) { + ERROR("Invalid boundary tag (received 0x%x\n",tempbt); + return -1; + } + temptag=palacios_alloc(taglen); + if (!temptag) { + ERROR("failed to allocate temptag\n"); + return -1; + } + + if (recv_msg(nks->ns,(char*)(&slen),sizeof(slen))!=sizeof(slen)) { + ERROR("Cannot receive tag len in read_key_net\n"); + palacios_free(temptag); + return -1; + } + + if (slen!=taglen) { + ERROR("Tag len expected does not matched tag len decoded in read_key_net\n"); + palacios_free(temptag); + return -1; + } + + if (recv_msg(nks->ns,temptag,taglen)!=taglen) { + ERROR("Cannot recieve tag in read_key_net\n"); + palacios_free(temptag); + return -1; + } + + if (memcmp(temptag,tag,taglen)) { + ERROR("Tag mismatch\n"); + palacios_free(temptag); + return -1; + } + if (recv_msg(nks->ns,(char*)(&slen),sizeof(slen))!=sizeof(slen)) { ERROR("Cannot receive data len in read_key_net\n"); + palacios_free(temptag); return -1; } if (slen!=len) { ERROR("Data len expected does not matched data len decoded in read_key_net\n"); + palacios_free(temptag); return -1; } if (recv_msg(nks->ns,buf,len)!=len) { ERROR("Cannot recieve data in read_key_net\n"); + palacios_free(temptag); return -1; } + + palacios_free(temptag); } else { ERROR("Permissions incorrect for the stream in read_key_net\n"); @@ -1961,8 +2923,10 @@ static v3_keyed_stream_t open_stream(char *url, return open_stream_file(url,ot); } else if (!strncasecmp(url,"user:",5)) { return open_stream_user(url,ot); - } else if(!strncasecmp(url,"net:",4)){ + } else if (!strncasecmp(url,"net:",4)){ return open_stream_net(url,ot); + } else if (!strncasecmp(url,"textfile:",9)) { + return open_stream_textfile(url,ot); } else { ERROR("unsupported type in attempt to open keyed stream \"%s\"\n",url); return 0; @@ -1979,6 +2943,9 @@ static void close_stream(v3_keyed_stream_t stream) case STREAM_FILE: return close_stream_file(stream); break; + case STREAM_TEXTFILE: + return close_stream_textfile(stream); + break; case STREAM_USER: return close_stream_user(stream); break; @@ -2003,6 +2970,9 @@ static void preallocate_hint_key(v3_keyed_stream_t stream, case STREAM_FILE: preallocate_hint_key_file(stream,key,size); break; + case STREAM_TEXTFILE: + preallocate_hint_key_textfile(stream,key,size); + break; case STREAM_USER: return preallocate_hint_key_user(stream,key,size); break; @@ -2028,6 +2998,9 @@ static v3_keyed_stream_key_t open_key(v3_keyed_stream_t stream, case STREAM_FILE: return open_key_file(stream,key); break; + case STREAM_TEXTFILE: + return open_key_textfile(stream,key); + break; case STREAM_USER: return open_key_user(stream,key); break; @@ -2053,6 +3026,9 @@ static void close_key(v3_keyed_stream_t stream, case STREAM_FILE: return close_key_file(stream,key); break; + case STREAM_TEXTFILE: + return close_key_textfile(stream,key); + break; case STREAM_USER: return close_key_user(stream,key); break; @@ -2069,22 +3045,27 @@ static void close_key(v3_keyed_stream_t stream, static sint64_t write_key(v3_keyed_stream_t stream, v3_keyed_stream_key_t key, + void *tag, + sint64_t taglen, void *buf, sint64_t len) { struct generic_keyed_stream *gks = (struct generic_keyed_stream *) stream; switch (gks->stype){ case STREAM_MEM: - return write_key_mem(stream,key,buf,len); + return write_key_mem(stream,key,tag,taglen,buf,len); break; case STREAM_FILE: - return write_key_file(stream,key,buf,len); + return write_key_file(stream,key,tag,taglen,buf,len); + break; + case STREAM_TEXTFILE: + return write_key_textfile(stream,key,tag,taglen,buf,len); break; case STREAM_USER: - return write_key_user(stream,key,buf,len); + return write_key_user(stream,key,tag,taglen,buf,len); break; case STREAM_NETWORK: - return write_key_net(stream,key,buf,len); + return write_key_net(stream,key,tag,taglen,buf,len); break; default: ERROR("unknown stream type %d in write_key\n",gks->stype); @@ -2097,22 +3078,27 @@ static sint64_t write_key(v3_keyed_stream_t stream, static sint64_t read_key(v3_keyed_stream_t stream, v3_keyed_stream_key_t key, + void *tag, + sint64_t taglen, void *buf, sint64_t len) { struct generic_keyed_stream *gks = (struct generic_keyed_stream *) stream; switch (gks->stype){ case STREAM_MEM: - return read_key_mem(stream,key,buf,len); + return read_key_mem(stream,key,tag,taglen,buf,len); break; case STREAM_FILE: - return read_key_file(stream,key,buf,len); + return read_key_file(stream,key,tag,taglen,buf,len); + break; + case STREAM_TEXTFILE: + return read_key_textfile(stream,key,tag,taglen,buf,len); break; case STREAM_USER: - return read_key_user(stream,key,buf,len); + return read_key_user(stream,key,tag,taglen,buf,len); break; case STREAM_NETWORK: - return read_key_net(stream,key,buf,len); + return read_key_net(stream,key,tag,taglen,buf,len); break; default: ERROR("unknown stream type %d in read_key\n",gks->stype); diff --git a/linux_usr/v3_user_keyed_stream_example.c b/linux_usr/v3_user_keyed_stream_example.c index 1886b73..6068b0c 100644 --- a/linux_usr/v3_user_keyed_stream_example.c +++ b/linux_usr/v3_user_keyed_stream_example.c @@ -30,9 +30,14 @@ int do_work(struct palacios_user_keyed_stream_op *req, // req->xfer : unused // req->user_key : the opaque key previously provided by you by an open key // req->buf_len : length of data - // req->buf : buffer (contains key name (open key) or value (write key)) - // - + // req->data_off : start of data within the buffer + // buf[0..data_off-1] is a tag (for read or write key) + // buf[data_off..buf_len-1] is data (for write key) + // req->buf : buffer + // open key: key name (open key) + // read key: tag + // write key: tag and data + // now built a response *resp = malloc(sizeof(struct palacios_user_keyed_stream_op) + datasize); if (!*resp) { @@ -42,6 +47,7 @@ int do_work(struct palacios_user_keyed_stream_op *req, (*resp)->len = sizeof(struct palacios_user_keyed_stream_op) + datasize; (*resp)->buf_len = datasize; + (*resp)->data_off = 0; // always the case for a response (*resp)->type = req->type; (*resp)->user_key = req->user_key; @@ -53,7 +59,10 @@ int do_work(struct palacios_user_keyed_stream_op *req, // resp->xfer : contains the size of data read or written (in read key or write key) // resp->user_key : unused // resp->buf_len : length of data following + // resp->data_off : offset of the data in the buffer // resp->buf : buffer (contains the data (read key)) + // read key: data + // write key: nothing return 0; diff --git a/linux_usr/v3_user_keyed_stream_file.c b/linux_usr/v3_user_keyed_stream_file.c index 21f1bf8..e7d99a9 100644 --- a/linux_usr/v3_user_keyed_stream_file.c +++ b/linux_usr/v3_user_keyed_stream_file.c @@ -94,6 +94,7 @@ int handle_open_key(struct palacios_user_keyed_stream_op *req, (*resp)->xfer=0; (*resp)->user_key=(void*)(uint64_t)fd; (*resp)->buf_len=0; + (*resp)->data_off=0; return 0; @@ -121,6 +122,7 @@ int handle_close_key(struct palacios_user_keyed_stream_op *req, (*resp)->xfer=rc; (*resp)->user_key=(void*)(uint64_t)fd; (*resp)->buf_len=0; + (*resp)->data_off=0; return 0; @@ -171,10 +173,32 @@ int handle_write_key(struct palacios_user_keyed_stream_op *req, { int fd; int rc; + sint64_t taglen = req->data_off; + sint64_t datalen = req->buf_len - req->data_off; + + if (datalen != req->xfer) { + fprintf(stderr,"Odd, xfer=%lld but datalen computed is %lld\n",req->xfer,datalen); + if (datalen > req->xfer) { + datalen = req->xfer; + } + } fd = (int) (uint64_t) (req->user_key); - rc = write_all(fd,req->buf,req->xfer); + // Write tag + rc = write_all(fd,req->buf,taglen); + + if (rc!=taglen) { + // failed to write tag, lets report as negative error + fprintf(stderr,"Failed to write tag (taglen=%lld, rc=%d)\n",taglen,rc); + rc = -1; + } else { + // Write data + rc = write_all(fd,req->buf+taglen,datalen); + if (rc!=datalen) { + fprintf(stderr,"Failed to write data (datalen=%lld, rc=%d)\n",datalen,rc); + } + } (*resp) = malloc(sizeof(struct palacios_user_keyed_stream_op)+0); @@ -187,6 +211,7 @@ int handle_write_key(struct palacios_user_keyed_stream_op *req, (*resp)->xfer=rc; (*resp)->user_key=(void*)(uint64_t)fd; (*resp)->buf_len=0; + (*resp)->data_off=0; return 0; @@ -199,6 +224,8 @@ int handle_read_key(struct palacios_user_keyed_stream_op *req, { int fd; int rc; + sint64_t taglen = req->data_off; + char temptag[taglen]; fd = (int) (uint64_t) (req->user_key); @@ -208,14 +235,31 @@ int handle_read_key(struct palacios_user_keyed_stream_op *req, return -1; } - rc = read_all(fd,(*resp)->buf,req->xfer); + // read key and compare + rc = read_all(fd,temptag,taglen); + + if (rc!=taglen) { + // Error + fprintf(stderr,"Failed to read tag (taglen=%lld, rc=%d)\n",taglen,rc); + rc = -1; + } else { + // tag check + if (memcmp(temptag,req->buf,taglen)) { + // tag mismatch + fprintf(stderr,"Tag mismatch in read tag\n"); + rc =-1; + } else { + // OK, do data read + rc = read_all(fd,(*resp)->buf,req->xfer); + } + } (*resp)->len=sizeof(struct palacios_user_keyed_stream_op) + (rc>0 ? rc : 0); (*resp)->type=req->type; (*resp)->xfer=rc; (*resp)->user_key=(void*)(uint64_t)fd; (*resp)->buf_len=rc>0 ? rc : 0; - + (*resp)->data_off = 0; return 0;