Hacky histogram generation; API's for client-side iterators.
git-svn-id: svn+ssh://svn.corp.yahoo.com/yahoo/yrl/labs/pnuts/code/logstore@656 8dad8b1f-cf64-0410-95b6-bcf113ffbcfe
This commit is contained in:
sears
parent
747f4df1f3
commit
1243d999a5
4 changed files with 118 additions and 32 deletions
|
|
@ -586,8 +586,63 @@ int op_stat_space_usage(pthread_data* data) {
|
|||
int op_stat_perf_report(pthread_data* data) {
|
||||
|
||||
}
|
||||
|
||||
//pageid_t diskTreeComponent::build_histogram(int xid, pageid_t bucket_count, const byte **key_array, size_t * size_array) {
|
||||
//
|
||||
//}
|
||||
|
||||
|
||||
int op_stat_histogram(pthread_data* data, size_t limit) {
|
||||
|
||||
if(limit < 3) {
|
||||
return writeoptosocket(*(data->workitem), LOGSTORE_PROTOCOL_ERROR);
|
||||
}
|
||||
|
||||
int xid = Tbegin();
|
||||
lladdIterator_t * it = diskTreeComponentIterator::open(xid, data->ltable->get_tree_c2()->get_root_rec());
|
||||
size_t count = 0;
|
||||
int err = 0;
|
||||
|
||||
while(diskTreeComponentIterator::next(xid, it)) { count++; }
|
||||
diskTreeComponentIterator::close(xid, it);
|
||||
|
||||
uint64_t stride;
|
||||
|
||||
if(count > limit) {
|
||||
stride = count / (limit-1);
|
||||
} else {
|
||||
stride = 1;
|
||||
}
|
||||
|
||||
datatuple * tup = datatuple::create(&stride, sizeof(stride));
|
||||
|
||||
if(!err) { err = writeoptosocket(*(data->workitem), LOGSTORE_RESPONSE_SENDING_TUPLES); }
|
||||
if(!err) { err = writetupletosocket(*(data->workitem), tup); }
|
||||
|
||||
datatuple::freetuple(tup);
|
||||
|
||||
size_t cur_stride = 0;
|
||||
size_t i = 0;
|
||||
it = diskTreeComponentIterator::open(xid, data->ltable->get_tree_c2()->get_root_rec());
|
||||
while(diskTreeComponentIterator::next(xid, it)) {
|
||||
i++;
|
||||
if(i == count || !cur_stride) { // do we want to send this key? (this matches the first, last and interior keys)
|
||||
byte * key;
|
||||
size_t keylen= diskTreeComponentIterator::key(xid, it, &key);
|
||||
tup = datatuple::create(key, keylen);
|
||||
|
||||
if(!err) { err = writetupletosocket(*(data->workitem), tup); }
|
||||
|
||||
datatuple::freetuple(tup);
|
||||
cur_stride = stride;
|
||||
}
|
||||
cur_stride--;
|
||||
}
|
||||
|
||||
diskTreeComponentIterator::close(xid, it);
|
||||
if(!err){ err = writeendofiteratortosocket(*(data->workitem)); }
|
||||
Tcommit(xid);
|
||||
return err;
|
||||
}
|
||||
int op_dbg_blockmap(pthread_data* data) {
|
||||
// produce a list of stasis regions
|
||||
|
|
@ -788,12 +843,8 @@ void * thread_work_fn( void * args)
|
|||
if(!err) { tuple = readtuplefromsocket(*(item->data->workitem), &err); }
|
||||
// read the second tuple from client
|
||||
if(!err) { tuple2 = readtuplefromsocket(*(item->data->workitem), &err); }
|
||||
|
||||
//step 3: process the tuple
|
||||
|
||||
// if(tuple) {
|
||||
// printf("Tuple req = %d key = >%s<\n", opcode, tuple->key());
|
||||
// }
|
||||
|
||||
if(!err) { err = dispatch_request(opcode, tuple, tuple2, item->data); }
|
||||
|
||||
//free the tuple
|
||||
|
|
|
|||
|
|
@ -63,10 +63,10 @@ static inline void close_conn(logstore_handle_t *l) {
|
|||
close(l->server_socket); //close the connection
|
||||
l->server_socket = -1;
|
||||
}
|
||||
datatuple *
|
||||
logstore_client_op(logstore_handle_t *l,
|
||||
uint8_t opcode, datatuple * tuple, datatuple * tuple2, uint64_t count)
|
||||
{
|
||||
|
||||
uint8_t
|
||||
logstore_client_op_returns_many(logstore_handle_t *l,
|
||||
uint8_t opcode, datatuple * tuple, datatuple * tuple2, uint64_t count) {
|
||||
|
||||
if(l->server_socket < 0)
|
||||
{
|
||||
|
|
@ -103,46 +103,73 @@ logstore_client_op(logstore_handle_t *l,
|
|||
DEBUG("sock opened %d\n", l->server_socket);
|
||||
}
|
||||
|
||||
|
||||
network_op_t err = 0;
|
||||
|
||||
//send the opcode
|
||||
if( writetosocket(l->server_socket, &opcode, sizeof(opcode)) ) { close_conn(l); return 0; }
|
||||
if( !err) { err = writetosocket(l->server_socket, &opcode, sizeof(opcode)); }
|
||||
|
||||
//send the first tuple
|
||||
if( writetupletosocket(l->server_socket, tuple) ) { close_conn(l); return 0; }
|
||||
if( !err) { err = writetupletosocket(l->server_socket, tuple); }
|
||||
|
||||
//send the second tuple
|
||||
if( writetupletosocket(l->server_socket, tuple2) ) { close_conn(l); return 0; }
|
||||
if( !err) { err = writetupletosocket(l->server_socket, tuple2); }
|
||||
|
||||
if( count != (uint64_t)-1) {
|
||||
if( writecounttosocket(l->server_socket, count) ) { close_conn(l); return 0; }
|
||||
if( (!err) && (count != (uint64_t)-1) ) {
|
||||
err = writecounttosocket(l->server_socket, count); }
|
||||
|
||||
network_op_t rcode = LOGSTORE_CONN_CLOSED_ERROR;
|
||||
if( !err) {
|
||||
rcode = readopfromsocket(l->server_socket,LOGSTORE_SERVER_RESPONSE);
|
||||
}
|
||||
|
||||
if( opiserror(rcode) ) { close_conn(l); }
|
||||
|
||||
network_op_t rcode = readopfromsocket(l->server_socket,LOGSTORE_SERVER_RESPONSE);
|
||||
return rcode;
|
||||
|
||||
if( opiserror(rcode) ) { close_conn(l); return 0; }
|
||||
}
|
||||
datatuple *
|
||||
logstore_client_next_tuple(logstore_handle_t *l) {
|
||||
assert(l->server_socket != -1); // otherwise, then the client forgot to check a return value...
|
||||
int err = 0;
|
||||
datatuple * ret = readtuplefromsocket(l->server_socket, &err);
|
||||
if(err) {
|
||||
close_conn(l);
|
||||
if(ret) {
|
||||
datatuple::freetuple(ret);
|
||||
ret = NULL;
|
||||
}
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
datatuple *
|
||||
logstore_client_op(logstore_handle_t *l,
|
||||
uint8_t opcode, datatuple * tuple, datatuple * tuple2, uint64_t count)
|
||||
{
|
||||
network_op_t rcode = logstore_client_op_returns_many(l, opcode, tuple, tuple2, count);
|
||||
|
||||
datatuple * ret = 0;
|
||||
if(opiserror(rcode)) { return NULL; }
|
||||
|
||||
datatuple * ret = NULL;
|
||||
|
||||
if(rcode == LOGSTORE_RESPONSE_SENDING_TUPLES)
|
||||
{ int err;
|
||||
uint64_t count = 0; // XXX
|
||||
datatuple *nxt;
|
||||
while(( nxt = readtuplefromsocket(l->server_socket, &err) )) {
|
||||
if(ret) datatuple::freetuple(ret); // XXX
|
||||
ret = nxt;
|
||||
if(err) { close_conn(l); return 0; }
|
||||
count++;
|
||||
}
|
||||
if(count > 1) { fprintf(stderr, "XXX return count: %lld but iterators are not handled by the logstore_client_op api\n", count); }
|
||||
{
|
||||
ret = logstore_client_next_tuple(l);
|
||||
if(ret) {
|
||||
datatuple *nxt = logstore_client_next_tuple(l);
|
||||
if(nxt) {
|
||||
fprintf(stderr, "Opcode %d returned multiple tuples, but caller expects zero or one. Closing connection.\n", (int)opcode);
|
||||
datatuple::freetuple(nxt);
|
||||
datatuple::freetuple(ret);
|
||||
close_conn(l);
|
||||
ret = 0;
|
||||
}
|
||||
}
|
||||
} else if(rcode == LOGSTORE_RESPONSE_SUCCESS) {
|
||||
ret = tuple ? tuple : datatuple::create("", 1);
|
||||
} else {
|
||||
assert(rcode == LOGSTORE_RESPONSE_FAIL); // if this is an invalid response, we should have noticed above
|
||||
ret = 0;
|
||||
}
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
|
|
|
|||
13
tcpclient.h
13
tcpclient.h
|
|
@ -15,9 +15,16 @@ typedef struct logstore_handle_t logstore_handle_t;
|
|||
logstore_handle_t * logstore_client_open(const char *host, int portnum, int timeout);
|
||||
|
||||
datatuple * logstore_client_op(logstore_handle_t* l,
|
||||
uint8_t opcode,
|
||||
datatuple *tuple = NULL, datatuple *tuple2 = NULL,
|
||||
uint64_t count = (uint64_t)-1);
|
||||
uint8_t opcode,
|
||||
datatuple *tuple = NULL, datatuple *tuple2 = NULL,
|
||||
uint64_t count = (uint64_t)-1);
|
||||
|
||||
uint8_t logstore_client_op_returns_many(logstore_handle_t *l,
|
||||
uint8_t opcode,
|
||||
datatuple * tuple = NULL, datatuple * tuple2 = NULL,
|
||||
uint64_t count = (uint64_t)-1);
|
||||
|
||||
datatuple * logstore_client_next_tuple(logstore_handle_t *l);
|
||||
|
||||
int logstore_client_close(logstore_handle_t* l);
|
||||
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
CREATE_CLIENT_EXECUTABLE(dump_blockmap)
|
||||
CREATE_CLIENT_EXECUTABLE(drop_database)
|
||||
CREATE_CLIENT_EXECUTABLE(space_usage)
|
||||
CREATE_CLIENT_EXECUTABLE(histogram)
|
||||
|
|
|
|||
Loading…
Reference in a new issue