| /* |
| * Copyright (C) 2011, 2014, 2015 Filip Pizlo. All rights reserved. |
| * Redistribution and use in source and binary forms, with or without |
| * modification, are permitted provided that the following conditions |
| * are met: |
| * 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. |
| * |
| * THIS SOFTWARE IS PROVIDED BY FILIP PIZLO ``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 FILIP PIZLO OR |
| * CONTRIBUTORS 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. |
| */ |
| |
| #include "tsf_internal.h" |
| #include "tsf_format.h" |
| #include "tsf_fsdb_protocol.h" |
| |
| #include <time.h> |
| #include <dirent.h> |
| #include <sys/types.h> |
| #include <stdio.h> |
| #include <sys/stat.h> |
| #include <sys/types.h> |
| #include <errno.h> |
| #include <unistd.h> |
| #include <fcntl.h> |
| |
| #ifdef HAVE_BSDSOCKS |
| #include <sys/types.h> |
| #include <sys/socket.h> |
| #include <sys/un.h> |
| #include <netinet/in.h> |
| #include <netdb.h> |
| #include <errno.h> |
| #include <sys/ioctl.h> |
| #include <arpa/inet.h> |
| #include <netinet/tcp.h> |
| #endif |
| |
| #ifdef HAVE_BSDSOCKS |
| static tsf_fsdb_response_t *connection_read(tsf_fsdb_connection_t *con); |
| |
| static tsf_bool_t open_connection(int domain, |
| int type, |
| int protocol, |
| const struct sockaddr *addr, |
| socklen_t addrlen, |
| int *fd) { |
| int flag; |
| |
| *fd=socket(domain,type,protocol); |
| if (*fd<0) { |
| tsf_set_errno("Could not create socket with domain %d, type %d, protocol %d", |
| domain,type,protocol); |
| return tsf_false; |
| } |
| |
| if (connect(*fd,addr,addrlen)<0) { |
| tsf_set_errno("Could not establish connection"); |
| close(*fd); |
| return tsf_false; |
| } |
| |
| flag=1; |
| if (setsockopt(*fd, IPPROTO_TCP, TCP_NODELAY, (char*)&flag, sizeof(flag))<0) { |
| tsf_set_errno("Could not disable Nagle's algorithm"); |
| close(*fd); |
| return tsf_false; |
| } |
| |
| return tsf_true; |
| } |
| |
| static tsf_fsdb_connection_t *connection_create(int fd, |
| tsf_fsdb_t *fsdb) { |
| tsf_fsdb_connection_t *result; |
| tsf_fsdb_response_t *rsp; |
| |
| tsf_assert(fd>=0); |
| |
| result=malloc(sizeof(tsf_fsdb_connection_t)); |
| if (result==NULL) { |
| tsf_set_errno("Could not malloc tsf_fsdb_connection_t"); |
| goto error1; |
| } |
| |
| result->buf_size=0; |
| result->buf_cursor=0; |
| result->buf=NULL; |
| |
| result->fd=fd; |
| |
| result->in=tsf_stream_file_input_fd_open( |
| fd,fsdb->limits,&fsdb->rdr_attr,tsf_false,TSF_DEF_NET_BUF_SIZE); |
| if (result->in==NULL) { |
| goto error2; |
| } |
| |
| result->out=tsf_stream_file_output_fd_open( |
| fd,&fsdb->wtr_attr,tsf_false,TSF_DEF_NET_BUF_SIZE); |
| if (result->out==NULL) { |
| goto error3; |
| } |
| |
| rsp=connection_read(result); |
| if (rsp==NULL) { |
| goto error4; |
| } |
| |
| if (rsp->payload.value!=tsf_fsdb_response__payload__welcome) { |
| tsf_set_error(TSF_E_PARSE_ERROR, |
| "Bad response from FSDB server: response %s with state %s", |
| tsf_fsdb_response__payload__to_str(rsp->payload.value), |
| tsf_fsdb_state__to_str(rsp->state.value)); |
| tsf_region_free(rsp); |
| goto error4; |
| } |
| |
| tsf_region_free(rsp); |
| |
| return result; |
| |
| error4: |
| tsf_stream_file_output_close(result->out); |
| error3: |
| tsf_stream_file_input_close(result->in); |
| error2: |
| free(result); |
| error1: |
| close(fd); |
| return NULL; |
| } |
| |
| static void connection_destroy(tsf_fsdb_connection_t *con) { |
| if (con->buf!=NULL) { |
| free(con->buf); |
| } |
| tsf_stream_file_input_close(con->in); |
| tsf_stream_file_output_close(con->out); |
| close(con->fd); |
| free(con); |
| } |
| |
| static tsf_fsdb_response_t *connection_read(tsf_fsdb_connection_t *con) { |
| tsf_fsdb_response_t *rsp=tsf_fsdb_response__read(con->in); |
| if (rsp==NULL) { |
| return NULL; |
| } |
| |
| con->clear=(rsp->state.value==tsf_fsdb_state__clear); |
| |
| return rsp; |
| } |
| |
| static void connection_return(tsf_fsdb_t *fsdb, |
| tsf_fsdb_connection_t *con) { |
| if (con->clear) { |
| tsf_mutex_lock(&fsdb->lock); |
| if (fsdb->u.remote.connection==NULL) { |
| fsdb->u.remote.connection=con; |
| con=NULL; |
| } |
| tsf_mutex_unlock(&fsdb->lock); |
| } |
| |
| if (con!=NULL) { |
| connection_destroy(con); |
| } |
| } |
| |
| static tsf_fsdb_connection_t *create_first_connection(tsf_fsdb_t *fsdb, |
| const char *hostname, |
| int port) { |
| int fd=-1; |
| tsf_bool_t connected=tsf_false; |
| |
| #ifdef HAVE_GETADDRINFO |
| struct addrinfo hints; |
| struct addrinfo *result,*cur; |
| char buf[32]; |
| |
| bzero(&hints,sizeof(hints)); |
| hints.ai_family=AF_INET; |
| hints.ai_socktype=SOCK_STREAM; |
| |
| snprintf(buf,sizeof(buf),"%d",port); |
| |
| if (getaddrinfo(hostname,buf,&hints,&result)!=0) { |
| tsf_set_error(TSF_E_HOSTNAME, |
| "Could not lookup hostname %s", |
| hostname); |
| return NULL; |
| } |
| |
| for (cur=result;cur!=NULL;cur=cur->ai_next) { |
| if (open_connection(cur->ai_family, |
| cur->ai_socktype, |
| cur->ai_protocol, |
| cur->ai_addr, |
| cur->ai_addrlen, |
| &fd)) { |
| connected=tsf_true; |
| fsdb->u.remote.result=result; |
| fsdb->u.remote.cur=cur; |
| break; |
| } |
| } |
| |
| if (!connected) { |
| freeaddrinfo(result); |
| } |
| #else |
| struct hostent *hp = gethostbyname(hostname); |
| |
| if (hp == NULL) { |
| tsf_set_error(TSF_E_HOSTNAME, |
| "Could not lookup hostname %s", |
| hostname); |
| return NULL; |
| } else { |
| unsigned i; |
| for (i=0;hp->h_addr_list[i]!=NULL;++i) { |
| struct sockaddr_in addr; |
| socklen_t slen=sizeof(addr); |
| |
| bzero(&addr,slen); |
| addr.sin_family=AF_INET; |
| addr.sin_port=htons(port); |
| addr.sin_addr.s_addr=*(struct in_addr*)(hp->h_addr_list[i]); |
| |
| if (open_connection(AF_INET, |
| SOCK_STREAM, |
| 0, |
| &addr, |
| slen, |
| &fd)) { |
| connected=tsf_true; |
| fsdb->u.remote.addr=*(uint32_t*)(hp->h_addr_list[i]); |
| fsdb->u.remote.port=port; |
| break; |
| } |
| } |
| } |
| #endif |
| |
| if (!connected) { |
| return NULL; |
| } |
| |
| return connection_create(fd,fsdb); |
| } |
| |
| static tsf_fsdb_connection_t *create_connection(tsf_fsdb_t *fsdb) { |
| int fd; |
| |
| #ifdef HAVE_GETADDRINFO |
| if (!open_connection(fsdb->u.remote.cur->ai_family, |
| fsdb->u.remote.cur->ai_socktype, |
| fsdb->u.remote.cur->ai_protocol, |
| fsdb->u.remote.cur->ai_addr, |
| fsdb->u.remote.cur->ai_addrlen, |
| &fd)) { |
| return NULL; |
| } |
| #else |
| struct sockaddr_in addr; |
| socklen_t slen=sizeof(addr); |
| |
| bzero(&addr,slen); |
| addr.sin_family=AF_INET; |
| addr.sin_port=htons(fsdb->u.remote.port); |
| addr.sin_addr.s_addr=*(struct in_addr*)(&fsdb->u.remote.addr); |
| |
| if (!open_connection(AF_INET, |
| SOCK_STREAM, |
| 0, |
| &addr, |
| slen, |
| &fd)) { |
| return NULL; |
| } |
| #endif |
| |
| return connection_create(fd,fsdb); |
| } |
| |
| static tsf_fsdb_response_t *first_action(tsf_fsdb_t *fsdb, |
| tsf_fsdb_connection_t **connection, |
| tsf_fsdb_command_t *cmd) { |
| unsigned i; |
| tsf_mutex_lock(&fsdb->lock); |
| *connection=fsdb->u.remote.connection; |
| fsdb->u.remote.connection=NULL; |
| tsf_mutex_unlock(&fsdb->lock); |
| |
| for (i=0;;++i) { |
| if (*connection!=NULL && |
| tsf_fsdb_command__write((*connection)->out,cmd) && |
| tsf_stream_file_output_flush((*connection)->out)) { |
| tsf_fsdb_response_t *response=connection_read(*connection); |
| if (response!=NULL) { |
| return response; |
| } |
| } |
| |
| if (i==0) { |
| /* retry - the connection may have gone stale */ |
| if (*connection!=NULL) { |
| connection_destroy(*connection); |
| } |
| |
| *connection=create_connection(fsdb); |
| if (*connection==NULL) { |
| return NULL; |
| } |
| } else { |
| /* already tried before, now fail permanently and destroy the connection. */ |
| tsf_assert(*connection!=NULL); |
| connection_destroy(*connection); |
| *connection=NULL; |
| return NULL; |
| } |
| } |
| } |
| |
| static void set_error(tsf_fsdb_response_t *rsp) { |
| switch (rsp->payload.value) { |
| case tsf_fsdb_response__payload__sys_error: |
| tsf_set_error(TSF_E_EXTERNAL, |
| "FSDB server responded with system error: %s", |
| rsp->payload.u.sys_error.msg); |
| break; |
| case tsf_fsdb_response__payload__proto_error: |
| tsf_set_error(TSF_E_EXTERNAL, |
| "FSDB server responded with protocol error: %s", |
| rsp->payload.u.proto_error.msg); |
| break; |
| case tsf_fsdb_response__payload__not_found: |
| tsf_set_error(TSF_E_ELE_NOT_FOUND, |
| "FSDB server could not find the requested element"); |
| break; |
| case tsf_fsdb_response__payload__in_eof: |
| tsf_set_error(TSF_E_EOF, |
| "End of file from FSDB server"); |
| break; |
| default: |
| tsf_set_error(TSF_E_PARSE_ERROR, |
| "Unexpected or unrecognized response from FSDB server; " |
| "state = %s, response = %s", |
| tsf_fsdb_state__to_str(rsp->state.value), |
| tsf_fsdb_response__payload__to_str(rsp->payload.value)); |
| break; |
| } |
| } |
| |
| static tsf_bool_t network_writer_flush(tsf_fsdb_out_t *out) { |
| tsf_fsdb_command_t cmd; |
| tsf_fsdb_response_t *rsp; |
| |
| cmd.value=tsf_fsdb_command__out_write; |
| cmd.u.out_write.data.data=out->u.remote.connection->buf; |
| cmd.u.out_write.data.len=out->u.remote.connection->buf_cursor; |
| if (!tsf_fsdb_command__write(out->u.remote.connection->out, |
| &cmd) || |
| !tsf_stream_file_output_flush(out->u.remote.connection->out)) { |
| return tsf_false; |
| } |
| |
| rsp=connection_read(out->u.remote.connection); |
| if (rsp==NULL) { |
| return tsf_false; |
| } |
| |
| if (rsp->state.value!=tsf_fsdb_state__put || |
| rsp->payload.value!=tsf_fsdb_response__payload__ok) { |
| set_error(rsp); |
| tsf_region_free(rsp); |
| return tsf_false; |
| } |
| |
| tsf_region_free(rsp); |
| |
| out->u.remote.connection->buf_cursor=0; |
| |
| return tsf_true; |
| } |
| |
| static tsf_bool_t network_writer(void *arg, |
| const void *buf_, |
| uint32_t len) { |
| tsf_fsdb_out_t *out=(tsf_fsdb_out_t*)arg; |
| uint8_t *buf = (uint8_t*)buf_; |
| uint32_t towrite; |
| |
| while (len>0) { |
| towrite=tsf_min(len, |
| out->u.remote.connection->buf_size- |
| out->u.remote.connection->buf_cursor); |
| memcpy(out->u.remote.connection->buf+ |
| out->u.remote.connection->buf_cursor, |
| buf, |
| towrite); |
| |
| out->u.remote.connection->buf_cursor+=towrite; |
| buf+=towrite; |
| len-=towrite; |
| |
| if (len==0) { |
| break; |
| } |
| |
| if (!network_writer_flush(out)) { |
| return tsf_false; |
| } |
| } |
| |
| return tsf_true; |
| } |
| |
| static tsf_bool_t network_reader(void *arg, |
| void *buf_, |
| uint32_t len) { |
| tsf_fsdb_in_t *in=(tsf_fsdb_in_t*)arg; |
| uint8_t *buf=(uint8_t*)buf_; |
| |
| while (len>0) { |
| if (in->u.remote.rsp!=NULL) { |
| uint32_t toread= |
| tsf_min(len, |
| in->u.remote.rsp->payload.u.in_data.data.len- |
| in->u.remote.rsp_cursor); |
| |
| memcpy(buf, |
| in->u.remote.rsp->payload.u.in_data.data.data+in->u.remote.rsp_cursor, |
| toread); |
| |
| in->u.remote.rsp_cursor+=toread; |
| len-=toread; |
| buf+=toread; |
| |
| if (len==0) { |
| break; |
| } |
| } |
| |
| if (in->u.remote.rsp!=NULL && |
| in->u.remote.rsp_cursor==in->u.remote.rsp->payload.u.in_data.data.len) { |
| |
| tsf_region_free(in->u.remote.rsp); |
| in->u.remote.rsp=NULL; |
| in->u.remote.rsp_cursor=0; |
| } |
| |
| if (in->u.remote.rsp==NULL) { |
| in->u.remote.rsp_cursor=0; |
| in->u.remote.rsp= |
| connection_read(in->u.remote.connection); |
| |
| switch (in->u.remote.rsp->payload.value) { |
| case tsf_fsdb_response__payload__in_data: |
| /* ok! */ |
| break; |
| case tsf_fsdb_response__payload__in_eof: |
| tsf_set_error((void*)buf==buf_? |
| TSF_E_UNEXPECTED_EOF:TSF_E_ERRONEOUS_EOF, |
| "EOF in response from FSDB server"); |
| tsf_region_free(in->u.remote.rsp); |
| in->u.remote.rsp=NULL; |
| return tsf_false; |
| default: |
| set_error(in->u.remote.rsp); |
| tsf_region_free(in->u.remote.rsp); |
| in->u.remote.rsp=NULL; |
| return tsf_false; |
| } |
| } |
| } |
| |
| return tsf_true; |
| } |
| #endif |
| |
| static void copy_params(tsf_fsdb_t *fsdb, |
| tsf_limits_t *limits, |
| tsf_zip_rdr_attr_t *rdr_attr, |
| tsf_zip_wtr_attr_t *wtr_attr) { |
| fsdb->limits=limits; |
| |
| if (rdr_attr==NULL) { |
| tsf_zip_rdr_attr_get_default(&fsdb->rdr_attr); |
| } else { |
| fsdb->rdr_attr=*rdr_attr; |
| } |
| |
| if (wtr_attr==NULL) { |
| tsf_zip_wtr_attr_get_default(&fsdb->wtr_attr); |
| } else { |
| fsdb->wtr_attr=*wtr_attr; |
| } |
| } |
| |
| tsf_fsdb_t *tsf_fsdb_open_local(const char *dirname, |
| tsf_limits_t *limits, |
| tsf_zip_rdr_attr_t *rdr_attr, |
| tsf_zip_wtr_attr_t *wtr_attr, |
| tsf_bool_t update, |
| tsf_bool_t truncate) { |
| tsf_fsdb_t *result; |
| char *marker_flnm; |
| FILE *fin; |
| char buf[9]; |
| |
| if (!update && truncate) { |
| tsf_set_error(TSF_E_INVALID_ARG, |
| "Cannot specify the truncate without the update " |
| "flag"); |
| return NULL; |
| } |
| |
| result=malloc(sizeof(tsf_fsdb_t)); |
| if (result==NULL) { |
| tsf_set_errno("Could not malloc tsf_fsdb_t"); |
| return NULL; |
| } |
| |
| result->kind=TSF_FSDB_LOCAL; |
| |
| result->u.local.dirname=strdup(dirname); |
| if (result->u.local.dirname==NULL) { |
| tsf_set_errno("Could not strdup dirname in tsf_fsdb_open_local"); |
| goto error1; |
| } |
| |
| if (update) { |
| if (mkdir(result->u.local.dirname,0777)<0) { |
| if (errno==EEXIST) { |
| /* all good. */ |
| } else { |
| tsf_set_errno("Could not create fsdb directory %s",result->u.local.dirname); |
| goto error2; |
| } |
| } else { |
| /* directory created; now create the marker file. */ |
| FILE *fout; |
| |
| marker_flnm=tsf_asprintf("%s/TSF_FSDB",result->u.local.dirname); |
| if (marker_flnm==NULL) { |
| tsf_set_errno("Could not allocate string"); |
| goto error2; |
| } |
| |
| fout=fopen(marker_flnm,"w"); |
| if (fout==NULL) { |
| tsf_set_errno("Could not open %s",marker_flnm); |
| free(marker_flnm); |
| goto error2; |
| } |
| |
| free(marker_flnm); |
| |
| fputs("TSF_FSDB",fout); /* FIXME: error checking? */ |
| |
| fclose(fout); |
| } |
| } |
| |
| marker_flnm=tsf_asprintf("%s/TSF_FSDB",result->u.local.dirname); |
| if (marker_flnm==NULL) { |
| tsf_set_errno("Could not allocate string"); |
| goto error2; |
| } |
| |
| fin=fopen(marker_flnm,"r"); |
| if (fin==NULL) { |
| tsf_set_errno("Could not open %s",marker_flnm); |
| free(marker_flnm); |
| goto error2; |
| } |
| |
| free(marker_flnm); |
| |
| bzero(buf,sizeof(buf)); |
| fgets(buf,sizeof(buf),fin); |
| |
| fclose(fin); |
| |
| if (strcmp(buf,"TSF_FSDB")) { |
| tsf_set_error(TSF_E_PARSE_ERROR, |
| "Bad FSDB database: the marker file contains %s " |
| "instead of TSF_FSDB", |
| buf); |
| goto error2; |
| } |
| |
| /* make sure that it exists and is a directory, and keep a handle to it |
| just for good measure. */ |
| |
| result->u.local.dirhandle=opendir(result->u.local.dirname); |
| if (result->u.local.dirhandle==NULL) { |
| tsf_set_errno("Could not open fsdb directory %s",result->u.local.dirname); |
| goto error2; |
| } |
| |
| /* if we're truncating, then delete all of the files. */ |
| if (truncate) { |
| struct dirent *entry; |
| |
| #if defined(HAVE_READDIR_R) && defined(HAVE_PATHCONF) |
| struct dirent *entrybuf; |
| long limit; |
| size_t len; |
| |
| limit=pathconf(result->u.local.dirname,_PC_PATH_MAX); |
| if (limit<0) { |
| tsf_set_errno("Could not get pathconf _PC_PATH_MAX for %s", |
| result->u.local.dirname); |
| goto error3; |
| } |
| |
| len=tsf_offsetof(struct dirent,d_name)+limit+1; |
| |
| entrybuf=malloc(len); |
| if (entrybuf==NULL) { |
| tsf_set_errno("Could not malloc struct dirent with size " fsz, |
| len); |
| goto error3; |
| } |
| #endif |
| |
| for (;;) { |
| char *flnm; |
| |
| #if defined(HAVE_READDIR_R) && defined(HAVE_PATHCONF) |
| entry=NULL; |
| if (readdir_r(result->u.local.dirhandle,entrybuf,&entry)!=0) { |
| tsf_set_errno("Could not readdir_r in %s", |
| result->u.local.dirname); |
| free(entrybuf); |
| goto error3; |
| } |
| #else |
| errno=0; |
| entry=readdir(result->u.local.dirhandle); |
| if (entry==NULL && errno!=0) { |
| tsf_set_errno("Could not readdir in %s", |
| result->u.local.dirname); |
| goto error3; |
| } |
| #endif |
| |
| if (entry==NULL) { |
| break; |
| } |
| |
| if (strcmp(entry->d_name,".") && |
| strcmp(entry->d_name,"..") && |
| strcmp(entry->d_name,"TSF_FSDB")) { |
| |
| flnm=tsf_asprintf("%s/%s",result->u.local.dirname,entry->d_name); |
| if (flnm==NULL) { |
| tsf_set_errno("Could not tsf_asprintf"); |
| #if defined(HAVE_READDIR_R) && defined(HAVE_PATHCONF) |
| free(entrybuf); |
| #endif |
| goto error3; |
| } |
| |
| if (unlink(flnm)<0) { |
| tsf_set_errno("Could not unlink %s",flnm); |
| #if defined(HAVE_READDIR_R) && defined(HAVE_PATHCONF) |
| free(entrybuf); |
| #endif |
| free(flnm); |
| goto error3; |
| } |
| |
| free(flnm); |
| } |
| } |
| |
| #if defined(HAVE_READDIR_R) && defined(HAVE_PATHCONF) |
| free(entrybuf); |
| #endif |
| } |
| |
| copy_params(result,limits,rdr_attr,wtr_attr); |
| |
| result->u.local.update=update; |
| result->u.local.truncate=truncate; |
| |
| result->u.local.create_cnt=(((int64_t)time(NULL))<<32)+(((int64_t)getpid())<<48); |
| |
| tsf_mutex_init(&result->lock); |
| |
| return result; |
| |
| error3: |
| closedir(result->u.local.dirhandle); |
| error2: |
| free((void*)result->u.local.dirname); |
| error1: |
| free(result); |
| return NULL; |
| } |
| |
| tsf_fsdb_t *tsf_fsdb_open_remote(const char *host, |
| int port, |
| tsf_limits_t *limits, |
| tsf_zip_rdr_attr_t *rdr_attr, |
| tsf_zip_wtr_attr_t *wtr_attr) { |
| #ifdef HAVE_BSDSOCKS |
| tsf_fsdb_t *result; |
| |
| result=malloc(sizeof(tsf_fsdb_t)); |
| if (result==NULL) { |
| tsf_set_errno("Could not malloc tsf_fsdb_t"); |
| return NULL; |
| } |
| |
| result->kind=TSF_FSDB_REMOTE; |
| |
| copy_params(result,limits,rdr_attr,wtr_attr); |
| |
| result->u.remote.connection=create_first_connection(result,host,port); |
| if (result->u.remote.connection==NULL) { |
| free(result); |
| return NULL; |
| } |
| |
| tsf_mutex_init(&result->lock); |
| |
| return result; |
| #else |
| tsf_set_error(TSF_E_NOT_SUPPORTED, |
| "BSD sockets support was not found on your system"); |
| return NULL; |
| #endif |
| } |
| |
| void tsf_fsdb_close(tsf_fsdb_t *fsdb) { |
| tsf_mutex_destroy(&fsdb->lock); |
| |
| switch (fsdb->kind) { |
| case TSF_FSDB_LOCAL: |
| closedir(fsdb->u.local.dirhandle); |
| free((void*)fsdb->u.local.dirname); |
| break; |
| #ifdef HAVE_BSDSOCKS |
| case TSF_FSDB_REMOTE: |
| #ifdef HAVE_GETADDRINFO |
| freeaddrinfo(fsdb->u.remote.result); |
| #endif |
| if (fsdb->u.remote.connection!=NULL) { |
| connection_destroy(fsdb->u.remote.connection); |
| } |
| break; |
| #endif |
| default: |
| tsf_abort("bad FSDB kind"); |
| } |
| free(fsdb); |
| } |
| |
| tsf_bool_t tsf_fsdb_guts_put(tsf_fsdb_t *fsdb, |
| tsf_buffer_t *key, |
| int *fd, |
| char **partflnm) { |
| tsf_stream_file_output_t *out; |
| |
| if (fsdb==NULL || key==NULL) { |
| return tsf_false; |
| } |
| |
| tsf_assert(fsdb->kind==TSF_FSDB_LOCAL); |
| |
| if (!fsdb->u.local.update) { |
| tsf_set_error(TSF_E_NOT_ALLOWED, |
| "Cannot put into FSDB because update flag was not set."); |
| return tsf_false; |
| } |
| |
| for (;;) { |
| uint64_t my_id; |
| |
| tsf_mutex_lock(&fsdb->lock); |
| my_id=fsdb->u.local.create_cnt; |
| fsdb->u.local.create_cnt+=(uint64_t)(uintptr_t)fd; |
| tsf_mutex_unlock(&fsdb->lock); |
| |
| *partflnm=tsf_asprintf("%s/part-" fui64,fsdb->u.local.dirname,my_id); |
| if (*partflnm==NULL) { |
| tsf_set_errno("Could not tsf_asprintf in tsf_fsdb_put"); |
| return tsf_false; |
| } |
| |
| *fd=open(*partflnm,O_CREAT|O_EXCL|O_WRONLY|O_TRUNC,0666); |
| if (*fd<0) { |
| if (errno==EEXIST) { |
| free(*partflnm); |
| continue; |
| } |
| tsf_set_errno("Could not open %s",*partflnm); |
| free(*partflnm); |
| return tsf_false; |
| } |
| |
| break; |
| } |
| |
| /* ok, now write the key as the first record of the file. */ |
| out=tsf_stream_file_output_fd_open(*fd,NULL,tsf_false,0); |
| if (out==NULL) { |
| goto abort; |
| } |
| |
| if (!tsf_stream_file_output_write(out,key)) { |
| tsf_stream_file_output_close(out); |
| goto abort; |
| } |
| |
| if (!tsf_stream_file_output_close(out)) { |
| goto abort; |
| } |
| |
| return tsf_true; |
| abort: |
| tsf_fsdb_guts_abort(fsdb,*partflnm); |
| free(*partflnm); |
| close(*fd); |
| return tsf_false; |
| } |
| |
| tsf_bool_t tsf_fsdb_guts_commit(tsf_fsdb_t *fsdb, |
| char *partflnm, |
| tsf_buffer_t *key, |
| char **dataflnm) { |
| char *name; |
| char *flnm; |
| uint32_t sha1sum[5]; |
| |
| if (fsdb==NULL || partflnm==NULL || key==NULL) { |
| return tsf_false; |
| } |
| |
| tsf_assert(fsdb->kind==TSF_FSDB_LOCAL); |
| |
| if (!fsdb->u.local.update) { |
| tsf_set_error(TSF_E_NOT_ALLOWED, |
| "Cannot commit into FSDB because update flag was not set."); |
| return tsf_false; |
| } |
| |
| if (!tsf_buffer_calc_sha1(key,sha1sum)) { |
| return tsf_false; |
| } |
| |
| name=tsf_sha1_sum_to_str(sha1sum); |
| if (name==NULL) { |
| return tsf_false; |
| } |
| |
| flnm=tsf_asprintf("%s/data-%s",fsdb->u.local.dirname,name); |
| if (flnm==NULL) { |
| tsf_set_errno("Could not asprintf in tsf_fsdb_guts_commit"); |
| free(name); |
| return tsf_false; |
| } |
| |
| free(name); |
| |
| if (rename(partflnm,flnm)<0) { |
| tsf_set_errno("Could not rename %s to %s", |
| partflnm,flnm); |
| free(flnm); |
| return tsf_false; |
| } |
| |
| if (dataflnm!=NULL) { |
| *dataflnm=flnm; |
| } else { |
| free(flnm); |
| } |
| |
| return tsf_true; |
| } |
| |
| tsf_bool_t tsf_fsdb_guts_abort(tsf_fsdb_t *fsdb, |
| char *partflnm) { |
| if (fsdb==NULL || partflnm==NULL) { |
| return tsf_false; |
| } |
| |
| tsf_assert(fsdb->kind==TSF_FSDB_LOCAL); |
| |
| if (!fsdb->u.local.update) { |
| tsf_set_error(TSF_E_NOT_ALLOWED, |
| "Cannot abort FSDB file because update flag was not set."); |
| return tsf_false; |
| } |
| |
| unlink(partflnm); |
| |
| return tsf_true; |
| } |
| |
| tsf_bool_t tsf_fsdb_guts_get(tsf_fsdb_t *fsdb, |
| tsf_buffer_t *key, |
| int *fd, |
| char **dataflnm) { |
| uint32_t sha1sum[5]; |
| char *name; |
| char *flnm; |
| tsf_serial_in_man_t *in_man; |
| tsf_buf_rdr_t *reader; |
| tsf_buffer_t *read_key; |
| uint32_t num_remaining; |
| |
| if (fsdb==NULL || key==NULL) { |
| /* propagate error */ |
| return tsf_false; |
| } |
| |
| tsf_assert(fsdb->kind==TSF_FSDB_LOCAL); |
| |
| if (!tsf_buffer_calc_sha1(key,sha1sum)) { |
| return tsf_false; |
| } |
| |
| name=tsf_sha1_sum_to_str(sha1sum); |
| if (name==NULL) { |
| return tsf_false; |
| } |
| |
| flnm=tsf_asprintf("%s/data-%s",fsdb->u.local.dirname,name); |
| if (flnm==NULL) { |
| tsf_set_errno("Could not asprintf in tsf_fsdb_get"); |
| free(name); |
| return tsf_false; |
| } |
| |
| free(name); |
| |
| *fd=open(flnm,O_RDONLY); |
| if (*fd<0) { |
| if (errno==ENOENT) { |
| tsf_set_error(TSF_E_ELE_NOT_FOUND, |
| "The requested key was not found in the FSDB database"); |
| } else { |
| tsf_set_errno("Could not open %s for reading",flnm); |
| } |
| free(flnm); |
| return tsf_false; |
| } |
| |
| if (dataflnm!=NULL) { |
| *dataflnm=flnm; |
| } else { |
| free(flnm); |
| } |
| |
| /* now verify that the key is what it is supposed to be. use a lower-level |
| approach to reading so we can control how much buffering is done. */ |
| reader=tsf_buf_reader_create(*fd,128,tsf_false); |
| if (reader==NULL) { |
| goto fail; |
| } |
| |
| in_man=tsf_serial_in_man_create(tsf_buf_reader_read,reader,fsdb->limits); |
| if (in_man==NULL) { |
| tsf_buf_reader_destroy(reader); |
| goto fail; |
| } |
| |
| read_key=tsf_serial_in_man_read(in_man); |
| if (read_key==NULL) { |
| tsf_serial_in_man_destroy(in_man); |
| tsf_buf_reader_destroy(reader); |
| goto fail; |
| } |
| |
| tsf_serial_in_man_destroy(in_man); |
| |
| num_remaining=tsf_buf_reader_get_remaining(reader); |
| tsf_buf_reader_destroy(reader); |
| |
| if (lseek(*fd,-((off_t)num_remaining),SEEK_CUR)<0) { |
| tsf_set_errno("Could not unseek buffered data"); |
| tsf_buffer_destroy(read_key); |
| goto fail; |
| } |
| |
| if (!tsf_buffer_compare(key,read_key)) { |
| tsf_set_error(TSF_E_PARSE_ERROR, |
| "Key mismatch on entry in FSDB database"); |
| tsf_buffer_destroy(read_key); |
| goto fail; |
| } |
| |
| tsf_buffer_destroy(read_key); |
| |
| return tsf_true; |
| |
| fail: |
| if (dataflnm!=NULL) { |
| free(*dataflnm); |
| } |
| close(*fd); |
| return tsf_false; |
| } |
| |
| tsf_bool_t tsf_fsdb_guts_rm(tsf_fsdb_t *fsdb, |
| tsf_buffer_t *key, |
| char **dataflnm) { |
| uint32_t sha1sum[5]; |
| char *name; |
| char *flnm; |
| |
| if (fsdb==NULL || key==NULL) { |
| /* propagate error */ |
| return tsf_false; |
| } |
| |
| tsf_assert(fsdb->kind==TSF_FSDB_LOCAL); |
| |
| if (!fsdb->u.local.update) { |
| tsf_set_error(TSF_E_NOT_ALLOWED, |
| "Cannot rm from FSDB because update flag was not set."); |
| return tsf_false; |
| } |
| |
| if (!tsf_buffer_calc_sha1(key,sha1sum)) { |
| return tsf_false; |
| } |
| |
| name=tsf_sha1_sum_to_str(sha1sum); |
| if (name==NULL) { |
| return tsf_false; |
| } |
| |
| flnm=tsf_asprintf("%s/data-%s",fsdb->u.local.dirname,name); |
| if (flnm==NULL) { |
| tsf_set_errno("Could not asprintf in tsf_fsdb_get"); |
| free(name); |
| return tsf_false; |
| } |
| |
| free(name); |
| |
| if (unlink(flnm)<0) { |
| if (errno==ENOENT) { |
| tsf_set_error(TSF_E_ELE_NOT_FOUND, |
| "The requested key was not found in the FSDB database"); |
| } else { |
| tsf_set_errno("Could not unlink %s",flnm); |
| } |
| free(flnm); |
| return tsf_false; |
| } |
| |
| if (dataflnm!=NULL) { |
| *dataflnm=flnm; |
| } else { |
| free(flnm); |
| } |
| return tsf_true; |
| } |
| |
| tsf_fsdb_out_t *tsf_fsdb_put(tsf_fsdb_t *fsdb, |
| tsf_buffer_t *key) { |
| if (fsdb==NULL || key==NULL) { |
| return NULL; |
| } |
| |
| switch (fsdb->kind) { |
| case TSF_FSDB_LOCAL: { |
| int fd; |
| char *flnm; |
| tsf_fsdb_out_t *result; |
| |
| if (!tsf_fsdb_guts_put(fsdb,key,&fd,&flnm)) { |
| return NULL; |
| } |
| |
| result=malloc(sizeof(tsf_fsdb_out_t)); |
| if (result==NULL) { |
| tsf_set_errno("Could not malloc in tsf_fsdb_put"); |
| close(fd); |
| tsf_fsdb_guts_abort(fsdb,flnm); |
| free(flnm); |
| return NULL; |
| } |
| |
| result->db=fsdb; |
| result->u.local.partname=flnm; |
| |
| result->u.local.out=tsf_stream_file_output_fd_open(fd, |
| &fsdb->wtr_attr, |
| tsf_true, |
| 0); |
| if (result->u.local.out==NULL) { |
| close(fd); |
| tsf_fsdb_guts_abort(fsdb,flnm); |
| free(flnm); |
| free(result); |
| return NULL; |
| } |
| |
| result->u.local.key=tsf_buffer_dup(key); |
| if (result->u.local.key==NULL) { |
| tsf_stream_file_output_close(result->u.local.out); |
| tsf_fsdb_guts_abort(fsdb,flnm); |
| free(flnm); |
| free(result); |
| return NULL; |
| } |
| |
| return result; |
| } |
| #ifdef HAVE_BSDSOCKS |
| case TSF_FSDB_REMOTE: { |
| tsf_fsdb_connection_t *connection; |
| tsf_fsdb_command_t cmd; |
| tsf_fsdb_response_t *rsp; |
| tsf_fsdb_out_t *result; |
| |
| cmd.value=tsf_fsdb_command__put; |
| tsf_assert(key!=NULL); |
| cmd.u.put.key=key; |
| |
| rsp=first_action(fsdb,&connection,&cmd); |
| if (rsp==NULL) { |
| return NULL; |
| } |
| |
| if (rsp->state.value!=tsf_fsdb_state__put || |
| rsp->payload.value!=tsf_fsdb_response__payload__ok) { |
| set_error(rsp); |
| connection_destroy(connection); |
| tsf_region_free(rsp); |
| return NULL; |
| } |
| |
| tsf_region_free(rsp); |
| |
| result=malloc(sizeof(tsf_fsdb_out_t)); |
| if (result==NULL) { |
| tsf_set_errno("Could not malloc in tsf_fsdb_put"); |
| connection_destroy(connection); |
| return NULL; |
| } |
| |
| result->db=fsdb; |
| result->u.remote.connection=connection; |
| |
| if (connection->buf==NULL) { |
| connection->buf_size=TSF_DEF_NET_BUF_SIZE; |
| connection->buf_cursor=0; |
| connection->buf=malloc(connection->buf_size); |
| if (connection->buf==NULL) { |
| connection_destroy(connection); |
| free(result); |
| return NULL; |
| } |
| } |
| |
| result->u.remote.out_man= |
| tsf_serial_out_man_create(network_writer, result); |
| if (result->u.remote.out_man==NULL) { |
| connection_destroy(connection); |
| free(result); |
| return NULL; |
| } |
| |
| return result; |
| } |
| #endif |
| default: tsf_abort("bad FSDB kind"); |
| } |
| } |
| |
| tsf_fsdb_in_t *tsf_fsdb_get(tsf_fsdb_t *fsdb, |
| tsf_buffer_t *key) { |
| if (fsdb==NULL || key==NULL) { |
| return NULL; |
| } |
| |
| switch (fsdb->kind) { |
| case TSF_FSDB_LOCAL: { |
| int fd; |
| tsf_fsdb_in_t *result; |
| |
| if (!tsf_fsdb_guts_get(fsdb,key,&fd,NULL)) { |
| return NULL; |
| } |
| |
| result=malloc(sizeof(tsf_fsdb_in_t)); |
| if (result==NULL) { |
| tsf_set_errno("Could not malloc in tsf_fsdb_get"); |
| close(fd); |
| return NULL; |
| } |
| |
| result->db=fsdb; |
| |
| result->u.local.in=tsf_stream_file_input_fd_open(fd, |
| fsdb->limits, |
| &fsdb->rdr_attr, |
| tsf_true, |
| 0); |
| if (result->u.local.in==NULL) { |
| close(fd); |
| free(result); |
| return NULL; |
| } |
| |
| result->u.local.key=tsf_buffer_dup(key); |
| if (result->u.local.key==NULL) { |
| tsf_stream_file_input_close(result->u.local.in); |
| free(result); |
| return NULL; |
| } |
| |
| return result; |
| } |
| #ifdef HAVE_BSDSOCKS |
| case TSF_FSDB_REMOTE: { |
| tsf_fsdb_connection_t *connection; |
| tsf_fsdb_command_t cmd; |
| tsf_fsdb_response_t *rsp; |
| tsf_fsdb_in_t *result; |
| |
| cmd.value=tsf_fsdb_command__get; |
| cmd.u.get.key=key; |
| |
| rsp=first_action(fsdb,&connection,&cmd); |
| if (rsp==NULL) { |
| return NULL; |
| } |
| |
| if (rsp->state.value!=tsf_fsdb_state__get || |
| rsp->payload.value!=tsf_fsdb_response__payload__ok) { |
| set_error(rsp); |
| if (rsp->payload.value==tsf_fsdb_response__payload__not_found) { |
| connection_return(fsdb,connection); |
| } else { |
| connection_destroy(connection); |
| } |
| tsf_region_free(rsp); |
| return NULL; |
| } |
| |
| tsf_region_free(rsp); |
| |
| result=malloc(sizeof(tsf_fsdb_in_t)); |
| if (result==NULL) { |
| tsf_set_errno("Could not malloc in tsf_fsdb_get"); |
| connection_destroy(connection); |
| return NULL; |
| } |
| |
| result->db=fsdb; |
| result->u.remote.connection=connection; |
| |
| result->u.remote.rsp=NULL; |
| result->u.remote.rsp_cursor=0; |
| |
| result->u.remote.in_man= |
| tsf_serial_in_man_create(network_reader, |
| result, |
| fsdb->limits); |
| if (result->u.remote.in_man==NULL) { |
| connection_destroy(connection); |
| free(result); |
| return NULL; |
| } |
| |
| return result; |
| } |
| #endif |
| default: tsf_abort("bad FSDB kind"); |
| } |
| } |
| |
| tsf_bool_t tsf_fsdb_rm(tsf_fsdb_t *fsdb, |
| tsf_buffer_t *key) { |
| if (fsdb==NULL || key==NULL) { |
| return tsf_false; |
| } |
| |
| switch (fsdb->kind) { |
| case TSF_FSDB_LOCAL: |
| return tsf_fsdb_guts_rm(fsdb,key,NULL); |
| #ifdef HAVE_BSDSOCKS |
| case TSF_FSDB_REMOTE: { |
| tsf_fsdb_connection_t *connection; |
| tsf_fsdb_command_t cmd; |
| tsf_fsdb_response_t *rsp; |
| |
| cmd.value=tsf_fsdb_command__rm; |
| cmd.u.rm.key=key; |
| |
| rsp=first_action(fsdb,&connection,&cmd); |
| if (rsp==NULL) { |
| return tsf_false; |
| } |
| |
| if (rsp->state.value!=tsf_fsdb_state__clear || |
| rsp->payload.value!=tsf_fsdb_response__payload__ok) { |
| set_error(rsp); |
| if (rsp->payload.value==tsf_fsdb_response__payload__not_found) { |
| connection_return(fsdb,connection); |
| } else { |
| connection_destroy(connection); |
| } |
| tsf_region_free(rsp); |
| return tsf_false; |
| } |
| |
| tsf_region_free(rsp); |
| connection_return(fsdb,connection); |
| |
| return tsf_true; |
| } |
| #endif |
| default: tsf_abort("bad FSDB kind"); |
| } |
| } |
| |
| void tsf_fsdb_in_close(tsf_fsdb_in_t *in) { |
| switch (in->db->kind) { |
| case TSF_FSDB_LOCAL: |
| tsf_stream_file_input_close(in->u.local.in); |
| tsf_buffer_destroy(in->u.local.key); |
| break; |
| #ifdef HAVE_BSDSOCKS |
| case TSF_FSDB_REMOTE: |
| connection_return(in->db,in->u.remote.connection); |
| if (in->u.remote.rsp!=NULL) { |
| tsf_region_free(in->u.remote.rsp); |
| } |
| tsf_serial_in_man_destroy(in->u.remote.in_man); |
| break; |
| #endif |
| default: tsf_abort("bad FSDB kind"); |
| } |
| free(in); |
| } |
| |
| tsf_buffer_t *tsf_fsdb_in_read(tsf_fsdb_in_t *in) { |
| switch (in->db->kind) { |
| case TSF_FSDB_LOCAL: |
| return tsf_stream_file_input_read(in->u.local.in); |
| #ifdef HAVE_BSDSOCKS |
| case TSF_FSDB_REMOTE: |
| return tsf_serial_in_man_read(in->u.remote.in_man); |
| #endif |
| default: tsf_abort("bad FSDB kind"); |
| } |
| } |
| |
| tsf_bool_t tsf_fsdb_out_commit(tsf_fsdb_out_t *out) { |
| if (out==NULL) { |
| return tsf_false; |
| } |
| |
| switch (out->db->kind) { |
| case TSF_FSDB_LOCAL: |
| if (!tsf_stream_file_output_close(out->u.local.out)) { |
| goto abort_local; |
| } |
| |
| if (!tsf_fsdb_guts_commit(out->db, |
| out->u.local.partname, |
| out->u.local.key, |
| NULL)) { |
| goto abort_local; |
| } |
| |
| free(out->u.local.partname); |
| tsf_buffer_destroy(out->u.local.key); |
| free(out); |
| |
| return tsf_true; |
| |
| abort_local: |
| tsf_fsdb_guts_abort(out->db,out->u.local.partname); |
| free(out->u.local.partname); |
| tsf_buffer_destroy(out->u.local.key); |
| free(out); |
| |
| return tsf_false; |
| #ifdef HAVE_BSDSOCKS |
| case TSF_FSDB_REMOTE: { |
| tsf_fsdb_command_t cmd; |
| tsf_fsdb_response_t *rsp; |
| |
| /* this part could be optimized - we're currently sending the commit command |
| * as a separate packet. it would be best to try to send it as part of the |
| * last packet of the flush. */ |
| |
| if (!network_writer_flush(out)) { |
| tsf_fsdb_out_abort(out); |
| return tsf_false; |
| } |
| |
| cmd.value=tsf_fsdb_command__out_commit; |
| if (!tsf_fsdb_command__write(out->u.remote.connection->out, |
| &cmd) || |
| !tsf_stream_file_output_flush(out->u.remote.connection->out)) { |
| tsf_fsdb_out_abort(out); |
| return tsf_false; |
| } |
| |
| rsp=connection_read(out->u.remote.connection); |
| if (rsp==NULL) { |
| tsf_fsdb_out_abort(out); |
| return tsf_false; |
| } |
| |
| if (rsp->state.value!=tsf_fsdb_state__clear || |
| rsp->payload.value!=tsf_fsdb_response__payload__ok) { |
| set_error(rsp); |
| tsf_region_free(rsp); |
| tsf_fsdb_out_abort(out); |
| return tsf_false; |
| } |
| |
| tsf_region_free(rsp); |
| |
| connection_return(out->db,out->u.remote.connection); |
| tsf_serial_out_man_destroy(out->u.remote.out_man); |
| free(out); |
| return tsf_true; |
| } |
| #endif |
| default: tsf_abort("bad FSDB kind"); |
| } |
| } |
| |
| void tsf_fsdb_out_abort(tsf_fsdb_out_t *out) { |
| switch (out->db->kind) { |
| case TSF_FSDB_LOCAL: |
| tsf_stream_file_output_close(out->u.local.out); |
| tsf_fsdb_guts_abort(out->db,out->u.local.partname); |
| tsf_buffer_destroy(out->u.local.key); |
| free(out->u.local.partname); |
| break; |
| #ifdef HAVE_BSDSOCKS |
| case TSF_FSDB_REMOTE: |
| connection_destroy(out->u.remote.connection); |
| tsf_serial_out_man_destroy(out->u.remote.out_man); |
| break; |
| #endif |
| default: tsf_abort("bad FSDB kind"); |
| } |
| free(out); |
| } |
| |
| tsf_bool_t tsf_fsdb_out_write(tsf_fsdb_out_t *out, |
| tsf_buffer_t *buf) { |
| if (out==NULL || buf==NULL) { |
| return tsf_false; |
| } |
| |
| switch (out->db->kind) { |
| case TSF_FSDB_LOCAL: |
| return tsf_stream_file_output_write(out->u.local.out,buf); |
| #ifdef HAVE_BSDSOCKS |
| case TSF_FSDB_REMOTE: |
| return tsf_serial_out_man_write(out->u.remote.out_man,buf); |
| #endif |
| default: tsf_abort("bad FSDB kind"); |
| } |
| } |
| |
| |
