optimizations

2 files changed, 416 insertions, 102 deletions

diff --git a/Makefile b/Makefile
index c2394d1..4741e6a 100644
--- a/Makefile
+++ b/Makefile
@@ -4,16 +4,12 @@ CFLAGS := -Wall -W -g
 LDLIBS := -lpthread -ldl -rdynamic
 
 PKGCONFIG := env PKG_CONFIG_PATH=/usr/local/lib/pkgconfig pkg-config
-FUSEVER := $(shell $(PKGCONFIG) --modversion fuse 2> /dev/null)
-ifeq ($(FUSEVER),)
-	LDLIBS += -lfuse
-else
-	CFLAGS += $(shell $(PKGCONFIG) --cflags fuse)
-	LDLIBS += $(shell $(PKGCONFIG) --libs fuse)
-endif
+CFLAGS += $(shell $(PKGCONFIG) --cflags fuse)
+LDLIBS += $(shell $(PKGCONFIG) --libs fuse)
+CFLAGS += $(shell $(PKGCONFIG) --cflags glib-2.0)
+LDLIBS += $(shell $(PKGCONFIG) --libs glib-2.0)
 
-CPPFLAGS := -D_FILE_OFFSET_BITS=64
-#CPPFLAGS += -DDEBUG
+CPPFLAGS := -D_FILE_OFFSET_BITS=64 -D_REENTRANT
 
 sshfs: sshfs.o 
 
diff --git a/sshfs.c b/sshfs.c
index 263e6b6..6439693 100644
--- a/sshfs.c
+++ b/sshfs.c
@@ -6,7 +6,13 @@
 #include <string.h>
 #include <stdint.h>
 #include <errno.h>
+#include <semaphore.h>
+#include <pthread.h>
+#include <netdb.h>
+#include <signal.h>
+#include <sys/time.h>
 #include <netinet/in.h>
+#include <glib.h>
 
 #define SSH_FXP_INIT                1
 #define SSH_FXP_VERSION             2
@@ -63,8 +69,14 @@
 
 #define MY_EOF 1
 
+#define MAX_REPLY_LEN (1 << 17)
+#define CACHE_TIMEOUT 20
+#define MAX_CACHE_SIZE 10000
+#define CACHE_CLEAN_INTERVAL 60
+
 static int infd;
 static int outfd;
+static int debug = 1;
 
 struct buffer {
     uint8_t *p;
@@ -72,6 +84,70 @@ struct buffer {
     size_t size;
 };
 
+struct request {
+    unsigned int want_reply;
+    sem_t ready;
+    uint8_t reply_type;
+    struct buffer reply;
+    struct timeval start;
+};
+
+struct openfile {
+    unsigned int read_ctr;
+    unsigned int write_ctr;
+    int rw;
+    struct buffer read_handle;
+    struct buffer write_handle;
+};
+
+struct node {
+    struct stat stat;
+    time_t updated;
+};
+
+static GHashTable *reqtab;
+static GHashTable *cache;
+static time_t last_cleaned;
+static pthread_mutex_t lock;
+static int processing_thread_started;
+
+#define DEBUG(format, args...) \
+	do { if (debug) fprintf(stderr, format, args); } while(0)
+
+static const char *type_name(uint8_t type)
+{
+    switch(type) {
+    case SSH_FXP_INIT:           return "INIT";
+    case SSH_FXP_VERSION:        return "VERSION";
+    case SSH_FXP_OPEN:           return "OPEN";
+    case SSH_FXP_CLOSE:          return "CLOSE";
+    case SSH_FXP_READ:           return "READ";
+    case SSH_FXP_WRITE:          return "WRITE";
+    case SSH_FXP_LSTAT:          return "LSTAT";
+    case SSH_FXP_FSTAT:          return "FSTAT";
+    case SSH_FXP_SETSTAT:        return "SETSTAT";
+    case SSH_FXP_FSETSTAT:       return "FSETSTAT";
+    case SSH_FXP_OPENDIR:        return "OPENDIR";
+    case SSH_FXP_READDIR:        return "READDIR";
+    case SSH_FXP_REMOVE:         return "REMOVE";
+    case SSH_FXP_MKDIR:          return "MKDIR";
+    case SSH_FXP_RMDIR:          return "RMDIR";
+    case SSH_FXP_REALPATH:       return "REALPATH";
+    case SSH_FXP_STAT:           return "STAT";
+    case SSH_FXP_RENAME:         return "RENAME";
+    case SSH_FXP_READLINK:       return "READLINK";
+    case SSH_FXP_SYMLINK:        return "SYMLINK";
+    case SSH_FXP_STATUS:         return "STATUS";
+    case SSH_FXP_HANDLE:         return "HANDLE";
+    case SSH_FXP_DATA:           return "DATA";
+    case SSH_FXP_NAME:           return "NAME";
+    case SSH_FXP_ATTRS:          return "ATTRS";
+    case SSH_FXP_EXTENDED:       return "EXTENDED";
+    case SSH_FXP_EXTENDED_REPLY: return "EXTENDED_REPLY";
+    default:                     return "???";
+    }
+}
+
 static inline void buf_init(struct buffer *buf, size_t size)
 {
     if (size) {
@@ -127,6 +203,11 @@ static inline void buf_add_mem(struct buffer *buf, const void *data,
     _buf_add_mem(buf, data, len);
 }
 
+static inline void buf_add_buf(struct buffer *buf, const struct buffer *bufa)
+{
+    _buf_add_mem(buf, bufa->p, bufa->len);
+}
+
 static inline void buf_add_uint8(struct buffer *buf, uint8_t val)
 {
     _buf_add_mem(buf, &val, 1);
@@ -277,8 +358,75 @@ static int buf_get_attrs(struct buffer *buf, struct stat *stbuf)
     return 0;
 }
 
+static int cache_clean_entry(void *_key, struct node *node, time_t *now)
+{
+    (void) _key;
+    if (*now > node->updated + CACHE_TIMEOUT)
+        return TRUE;
+    else
+        return FALSE;
+}
+
+static void cache_clean(void)
+{
+    time_t now = time(NULL);
+    if (g_hash_table_size(cache) > MAX_CACHE_SIZE ||
+        now > last_cleaned + CACHE_CLEAN_INTERVAL) {
+        g_hash_table_foreach_remove(cache, (GHRFunc) cache_clean_entry, &now);
+        last_cleaned = now;
+    }
+}
+
+static struct node *cache_lookup(const char *path)
+{
+    return g_hash_table_lookup(cache, path);    
+}
+
+static void cache_remove(const char *path)
+{
+    pthread_mutex_lock(&lock);
+    g_hash_table_remove(cache, path);
+    pthread_mutex_unlock(&lock);
+}
+
+static void cache_invalidate(const char *path)
+{
+    cache_remove(path);
+}
+
+static void cache_rename(const char *from, const char *to)
+{
+    cache_remove(from);
+    cache_remove(to);
+}
+
+static struct node *cache_get(const char *path)
+{
+    struct node *node = cache_lookup(path);
+    if (node == NULL) {
+        char *pathcopy = g_strdup(path);
+        node = g_new0(struct node, 1);
+        g_hash_table_insert(cache, pathcopy, node);
+    }
+    return node;
+}
+
+static void cache_add_attr(const char *path, const struct stat *stbuf)
+{
+    struct node *node;
+    time_t now;
+
+    pthread_mutex_lock(&lock);
+    node = cache_get(path);
+    now = time(NULL);
+    node->stat = *stbuf;
+    node->updated = time(NULL);
+    cache_clean();
+    pthread_mutex_unlock(&lock);
+}
+
 static int buf_get_entries(struct buffer *buf, fuse_dirh_t h,
-                           fuse_dirfil_t filler)
+                           fuse_dirfil_t filler, const char *path)
 {
     uint32_t count;
     unsigned i;
@@ -296,7 +444,11 @@ static int buf_get_entries(struct buffer *buf, fuse_dirh_t h,
         if (buf_get_string(buf, &longname) != -1) {
             free(longname);
             if (buf_get_attrs(buf, &stbuf) != -1) {
-                filler(h, name, stbuf.st_mode >> 12);
+                char *fullpath;
+                filler(h, name, stbuf.st_mode >> 12, 0);
+                fullpath = g_strdup_printf("%s/%s", path, name);
+                cache_add_attr(fullpath, &stbuf);
+                g_free(fullpath);
                 err = 0;
             }
         }
@@ -342,6 +494,45 @@ static int start_ssh(const char *host)
     return 0;
 }
 
+static int connect_to(char *host)
+{
+    int err;
+    int sock;
+    struct addrinfo *ai;
+    struct addrinfo hint;
+    char *port = strchr(host, ':');
+    if (port == NULL) {
+        fprintf(stderr, "destination format must be: `.host:port'\n");
+        return -1;
+    }
+    *port++ = '\0';
+
+    memset(&hint, 0, sizeof(hint));
+    hint.ai_family = PF_INET;
+    hint.ai_socktype = SOCK_STREAM;
+    err = getaddrinfo(host, port, &hint, &ai);
+    if (err) {
+        fprintf(stderr, "failed to resolve %s:%s: %s\n", host, port,
+                gai_strerror(err));
+        return -1;
+    }
+    sock = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol);
+    if (sock == -1) {
+        perror("failed to create socket");
+        return -1;
+    }
+    err = connect(sock, ai->ai_addr, ai->ai_addrlen);
+    if (err == -1) {
+        perror("failed to connect");
+        return -1;
+    }
+    freeaddrinfo(ai);
+
+    infd = sock;
+    outfd = sock;
+    return 0;
+}
+
 static int do_write(struct buffer *buf)
 {
     uint8_t *p = buf->p;
@@ -375,9 +566,11 @@ static int sftp_send(uint8_t type, struct buffer *buf)
     buf_init(&buf2, 5);
     buf_add_uint32(&buf2, buf->len + 1);
     buf_add_uint8(&buf2, type);
+    pthread_mutex_lock(&lock);
     res = do_write(&buf2);
     if (res != -1)
         res = do_write(buf);
+    pthread_mutex_unlock(&lock);
     buf_free(&buf2);
     return res;
 }
@@ -411,6 +604,10 @@ static int sftp_read(uint8_t *type, struct buffer *buf)
     res = do_read(&buf2);
     if (res != -1) {
         buf_get_uint32(&buf2, &len);
+        if (len > MAX_REPLY_LEN) {
+            fprintf(stderr, "reply len too large: %u\n", len);
+            return -1;
+        }
         buf_get_uint8(&buf2, type);
         buf_init(buf, len - 1);
         res = do_read(buf);
@@ -419,39 +616,117 @@ static int sftp_read(uint8_t *type, struct buffer *buf)
     return res;
 }
 
+static void *process_requests(void *_data)
+{
+    (void) _data;
+
+    while (1) {
+        int res;
+        struct buffer buf;
+        uint8_t type;
+        struct request *req;
+        uint32_t id;
+
+        buf_init(&buf, 0);
+        res = sftp_read(&type, &buf);
+        if (res == -1)
+            break;
+        if (buf_get_uint32(&buf, &id) == -1)
+            break;
+
+        pthread_mutex_lock(&lock);
+        req = g_hash_table_lookup(reqtab, (gpointer) id);
+        if (req == NULL)
+            fprintf(stderr, "request %i not found\n", id);
+        else
+            g_hash_table_remove(reqtab, (gpointer) id);
+        pthread_mutex_unlock(&lock);
+        if (req != NULL) {
+            struct timeval now;
+            unsigned int difftime;
+            gettimeofday(&now, NULL);
+            difftime = (now.tv_sec - req->start.tv_sec) * 1000;
+            difftime += (now.tv_usec - req->start.tv_usec) / 1000;
+            DEBUG("  [%05i] %14s %8ibytes (%ims)\n", id, type_name(type),
+                  buf.size+5, difftime);
+            req->reply = buf;
+            if (req->want_reply) {
+                req->reply_type = type;
+                sem_post(&req->ready);
+            } else {
+                buf_free(&req->reply);
+                sem_destroy(&req->ready);
+                free(req);
+            }
+        } else
+            buf_free(&buf);
+    }
+    kill(getpid(), SIGTERM);
+    return NULL;
+}
+
+static int start_processing_thread(void)
+{
+    int err;
+    pthread_t thread_id;
+    if (processing_thread_started)
+        return 0;
+
+    err = pthread_create(&thread_id, NULL, process_requests, NULL);
+    if (err) {
+        fprintf(stderr, "failed to create thread: %s\n", strerror(err));
+        return -EPERM;
+    }
+    pthread_detach(thread_id);
+    processing_thread_started = 1;
+    return 0;
+}
+
 static int sftp_request(uint8_t type, const struct buffer *buf,
                         uint8_t expect_type, struct buffer *outbuf)
 {
     int err;
     struct buffer buf2;
     uint32_t id = sftp_get_id();
-    uint32_t idback;
-    uint8_t typeback;
+    struct request *req = (struct request *) malloc(sizeof(struct request));
 
     buf_init(&buf2, buf->len + 4);
     buf_add_uint32(&buf2, id);
     buf_add_mem(&buf2, buf->p, buf->len);
+
+    req->want_reply = expect_type != 0 ? 1 : 0;
+    sem_init(&req->ready, 0, 0);
+    buf_init(&req->reply, 0);
+    pthread_mutex_lock(&lock);
+    err = start_processing_thread();
+    g_hash_table_insert(reqtab, (gpointer) id, req);
+    gettimeofday(&req->start, NULL);
+    DEBUG("[%05i] %s\n", id, type_name(type));
+    pthread_mutex_unlock(&lock);
+    if (err)
+        goto out;
     
     err = -EIO;
-    if (sftp_send(type, &buf2) == -1)
-        goto out;
-    buf_clear(&buf2);
-    if (sftp_read(&typeback, &buf2) == -1)
+    if (sftp_send(type, &buf2) == -1) {
+        pthread_mutex_lock(&lock);
+        g_hash_table_remove(reqtab, (gpointer) id);
+        pthread_mutex_unlock(&lock);
         goto out;
+    }
+    if (expect_type == 0) {
+        buf_free(&buf2);
+        return 0;
+    }
+
+    sem_wait(&req->ready);
     err = -EPROTO;
-    if (typeback != expect_type && typeback != SSH_FXP_STATUS) {
+    if (req->reply_type != expect_type && req->reply_type != SSH_FXP_STATUS) {
         fprintf(stderr, "protocol error\n");
         goto out;
     }
-    if (buf_get_uint32(&buf2, &idback) == -1)
-        goto out;
-    if (idback != id) {
-        fprintf(stderr, "Invalid ID received\n");
-        goto out;
-    }
-    if (typeback == SSH_FXP_STATUS) {
+    if (req->reply_type == SSH_FXP_STATUS) {
         uint32_t serr;
-        if (buf_get_uint32(&buf2, &serr) == -1)
+        if (buf_get_uint32(&req->reply, &serr) == -1)
             goto out;
         
         switch (serr) {
@@ -476,18 +751,21 @@ static int sftp_request(uint8_t type, const struct buffer *buf,
         default:                       err = -EPROTO; break;
         }
     } else {
-        buf_init(outbuf, buf2.size - buf2.len);
-        buf_get_mem(&buf2, outbuf->p, outbuf->size);
+        buf_init(outbuf, req->reply.size - req->reply.len);
+        buf_get_mem(&req->reply, outbuf->p, outbuf->size);
         err = 0;
     }
 
  out:
     buf_free(&buf2);
+    buf_free(&req->reply);
+    sem_destroy(&req->ready);
+    free(req);
     return err;
         
 }
 
-static int sshfs_getattr(const char *path, struct stat *stbuf)
+static int sshfs_send_getattr(const char *path, struct stat *stbuf)
 {
     int err;
     struct buffer buf;
@@ -501,9 +779,29 @@ static int sshfs_getattr(const char *path, struct stat *stbuf)
         buf_free(&outbuf);
     }
     buf_free(&buf);
+    if (!err)
+        cache_add_attr(path, stbuf);
     return err;
 }
 
+static int sshfs_getattr(const char *path, struct stat *stbuf)
+{
+    struct node *node;
+
+    pthread_mutex_lock(&lock);
+    node = cache_lookup(path);
+    if (node != NULL) {
+        time_t now = time(NULL);
+        if (now - node->updated < CACHE_TIMEOUT) {
+            *stbuf = node->stat;
+            pthread_mutex_unlock(&lock);
+            return 0;
+        }
+    }
+    pthread_mutex_unlock(&lock);
+    return sshfs_send_getattr(path, stbuf);
+}
+
 static int sshfs_readlink(const char *path, char *linkbuf, size_t size)
 {
     int err;
@@ -544,7 +842,7 @@ static int sshfs_getdir(const char *path, fuse_dirh_t h, fuse_dirfil_t filler)
             struct buffer name;
             err = sftp_request(SSH_FXP_READDIR, &handle, SSH_FXP_NAME, &name);
             if (!err) {
-                if (buf_get_entries(&name, h, filler) == -1)
+                if (buf_get_entries(&name, h, filler, path) == -1)
                     err = -EPROTO;
                 buf_free(&name);
             }
@@ -553,7 +851,7 @@ static int sshfs_getdir(const char *path, fuse_dirh_t h, fuse_dirfil_t filler)
             err = 0;
         
         
-        err2 = sftp_request(SSH_FXP_CLOSE, &handle, SSH_FXP_STATUS, NULL);
+        err2 = sftp_request(SSH_FXP_CLOSE, &handle, 0, NULL);
         if (!err)
             err = err2;
         buf_free(&handle);
@@ -579,7 +877,6 @@ static int sshfs_mknod(const char *path, mode_t mode, dev_t rdev)
     int err;
     struct buffer buf;
     struct buffer handle;
-
     (void) rdev;
 
     if ((mode & S_IFMT) != S_IFREG)
@@ -607,9 +904,9 @@ static int sshfs_symlink(const char *from, const char *to)
 {
     int err;
     struct buffer buf;
-    buf_init(&buf, 0);
     /* openssh sftp server doesn't follow standard: link target and
        link name are mixed up, so we must also be non-standard :( */
+    buf_init(&buf, 0);
     buf_add_string(&buf, from);
     buf_add_string(&buf, to);
     err = sftp_request(SSH_FXP_SYMLINK, &buf, SSH_FXP_STATUS, NULL);
@@ -624,6 +921,8 @@ static int sshfs_unlink(const char *path)
     buf_init(&buf, 0);
     buf_add_string(&buf, path);
     err = sftp_request(SSH_FXP_REMOVE, &buf, SSH_FXP_STATUS, NULL);
+    if (!err)
+        cache_remove(path);
     buf_free(&buf);
     return err;
 }
@@ -635,6 +934,8 @@ static int sshfs_rmdir(const char *path)
     buf_init(&buf, 0);
     buf_add_string(&buf, path);
     err = sftp_request(SSH_FXP_RMDIR, &buf, SSH_FXP_STATUS, NULL);
+    if (!err)
+        cache_remove(path);
     buf_free(&buf);
     return err;
 }
@@ -647,6 +948,8 @@ static int sshfs_rename(const char *from, const char *to)
     buf_add_string(&buf, from);
     buf_add_string(&buf, to);
     err = sftp_request(SSH_FXP_RENAME, &buf, SSH_FXP_STATUS, NULL);
+    if (!err)
+        cache_rename(from, to);
     buf_free(&buf);
     return err;
 }
@@ -660,6 +963,8 @@ static int sshfs_chmod(const char *path, mode_t mode)
     buf_add_uint32(&buf, SSH_FILEXFER_ATTR_PERMISSIONS);
     buf_add_uint32(&buf, mode);
     err = sftp_request(SSH_FXP_SETSTAT, &buf, SSH_FXP_STATUS, NULL);
+    if (!err)
+        cache_invalidate(path);
     buf_free(&buf);
     return err;
 }
@@ -687,6 +992,8 @@ static int sshfs_truncate(const char *path, off_t size)
     buf_add_uint32(&buf, SSH_FILEXFER_ATTR_SIZE);
     buf_add_uint64(&buf, size);
     err = sftp_request(SSH_FXP_SETSTAT, &buf, SSH_FXP_STATUS, NULL);
+    if (!err)
+        cache_invalidate(path);
     buf_free(&buf);
     return err;
 }
@@ -695,112 +1002,106 @@ static int sshfs_utime(const char *path, struct utimbuf *ubuf)
 {
     int err;
     struct buffer buf;
+    cache_remove(path);
     buf_init(&buf, 0);
     buf_add_string(&buf, path);
     buf_add_uint32(&buf, SSH_FILEXFER_ATTR_ACMODTIME);
     buf_add_uint32(&buf, ubuf->actime);
     buf_add_uint32(&buf, ubuf->modtime);
     err = sftp_request(SSH_FXP_SETSTAT, &buf, SSH_FXP_STATUS, NULL);
+    if (!err)
+        cache_invalidate(path);
     buf_free(&buf);
     return err;
 }
 
-static int sshfs_open(const char *path, int flags)
+static int sshfs_open(const char *path, struct fuse_file_info *fi)
 {
     int err;
     struct buffer buf;
-    struct buffer handle;
-    uint32_t pflags;
-    if ((flags & O_ACCMODE) == O_RDONLY)
+    struct buffer *handle;
+    uint32_t pflags = 0;
+    if ((fi->flags & O_ACCMODE) == O_RDONLY)
         pflags = SSH_FXF_READ;
-    else if((flags & O_ACCMODE) == O_WRONLY)
+    else if((fi->flags & O_ACCMODE) == O_WRONLY)
         pflags = SSH_FXF_WRITE;
-    else 
+    else if ((fi->flags & O_ACCMODE) == O_RDWR)
         pflags = SSH_FXF_READ | SSH_FXF_WRITE;
-     
+    else
+        return -EINVAL;
+
+    handle = g_new0(struct buffer, 1);
     buf_init(&buf, 0);
     buf_add_string(&buf, path);
     buf_add_uint32(&buf, pflags);
     buf_add_uint32(&buf, 0);
-    err = sftp_request(SSH_FXP_OPEN, &buf, SSH_FXP_HANDLE, &handle);
+    err = sftp_request(SSH_FXP_OPEN, &buf, SSH_FXP_HANDLE, handle);
     if (!err) {
-        int err2;
-        buf_finish(&handle);
-        err2 = sftp_request(SSH_FXP_CLOSE, &handle, SSH_FXP_STATUS, NULL);
-        if (!err)
-            err = err2;
-        buf_free(&handle);
-    }
+        buf_finish(handle);
+        fi->fh = (unsigned long) handle;
+    } else
+        g_free(handle);
     buf_free(&buf);
     return err;
 }
 
-static int sshfs_read(const char *path, char *rbuf, size_t size, off_t offset)
+static int sshfs_release(const char *path, struct fuse_file_info *fi)
+{
+    struct buffer *handle = (struct buffer *) fi->fh;
+    (void) path;
+    sftp_request(SSH_FXP_CLOSE, handle, 0, NULL);
+    buf_free(handle);
+    g_free(handle);
+    return 0;
+}
+
+static int sshfs_read(const char *path, char *rbuf, size_t size, off_t offset,
+                      struct fuse_file_info *fi)
 {
     int err;
     struct buffer buf;
-    struct buffer handle;
+    struct buffer data;
+    struct buffer *handle = (struct buffer *) fi->fh;
+    (void) path;
     buf_init(&buf, 0);
-    buf_add_string(&buf, path);
-    buf_add_uint32(&buf, SSH_FXF_READ);
-    buf_add_uint32(&buf, 0);
-    err = sftp_request(SSH_FXP_OPEN, &buf, SSH_FXP_HANDLE, &handle);
+    buf_add_buf(&buf, handle);
+    buf_add_uint64(&buf, offset);
+    buf_add_uint32(&buf, size);
+    err = sftp_request(SSH_FXP_READ, &buf, SSH_FXP_DATA, &data);
     if (!err) {
-        struct buffer data;
-        int err2;
-        buf_finish(&handle);
-        buf_add_uint64(&handle, offset);
-        buf_add_uint32(&handle, size);
-        err = sftp_request(SSH_FXP_READ, &handle, SSH_FXP_DATA, &data);
-        if (!err) {
-            uint32_t retsize;
-            err = -EPROTO;
-            if (buf_get_uint32(&data, &retsize) != -1) {
-                if (retsize > size)
-                    fprintf(stderr, "long read\n");
-                else {
-                    buf_get_mem(&data, rbuf, retsize);
-                    err = retsize;
-                }
+        uint32_t retsize;
+        err = -EPROTO;
+        if (buf_get_uint32(&data, &retsize) != -1) {
+            if (retsize > size)
+                fprintf(stderr, "long read\n");
+            else {
+                buf_get_mem(&data, rbuf, retsize);
+                err = retsize;
             }
-            buf_free(&data);
         }
-        err2 = sftp_request(SSH_FXP_CLOSE, &handle, SSH_FXP_STATUS, NULL);
-        if (err2 && err >= 0)
-            err = err2;
-        buf_free(&handle);
+        buf_free(&data);
     }
     buf_free(&buf);
     return err;    
 }
 
 static int sshfs_write(const char *path, const char *wbuf, size_t size,
-                       off_t offset)
+                       off_t offset, struct fuse_file_info *fi)
 {
     int err;
     struct buffer buf;
-    struct buffer handle;
+    struct buffer data;
+    struct buffer *handle = (struct buffer *) fi->fh;
+    (void) path;
+    data.p = (uint8_t *) wbuf;
+    data.len = size;
     buf_init(&buf, 0);
-    buf_add_string(&buf, path);
-    buf_add_uint32(&buf, SSH_FXF_WRITE);
-    buf_add_uint32(&buf, 0);
-    err = sftp_request(SSH_FXP_OPEN, &buf, SSH_FXP_HANDLE, &handle);
-    if (!err) {
-        struct buffer data;
-        int err2;
-        data.p = (uint8_t *) wbuf;
-        data.len = size;
-        buf_finish(&handle);
-        buf_add_uint64(&handle, offset);
-        buf_add_data(&handle, &data);
-        err = sftp_request(SSH_FXP_WRITE, &handle, SSH_FXP_STATUS, NULL);
-        err2 = sftp_request(SSH_FXP_CLOSE, &handle, SSH_FXP_STATUS, NULL);
-        if (err2 && err >= 0)
-            err = err2;
-        buf_free(&handle);
-    }
+    buf_add_buf(&buf, handle);
+    buf_add_uint64(&buf, offset);
+    buf_add_data(&buf, &data);
+    err = sftp_request(SSH_FXP_WRITE, &buf, SSH_FXP_STATUS, NULL);
     buf_free(&buf);
-    return err;    
+    return err ? err : (int) size;
 }
 
 static int sftp_init()
@@ -833,6 +1134,18 @@ static int sftp_init()
     return res;
 }
 
+static int processing_init(void)
+{
+    pthread_mutex_init(&lock, NULL);
+    reqtab = g_hash_table_new(NULL, NULL);
+    cache = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, g_free);
+    if (!reqtab || !cache) {
+        fprintf(stderr, "failed to create hash tables\n");
+        return -1;
+    }
+    return 0;
+}
+
 static struct fuse_operations sshfs_oper = {
     .getattr	= sshfs_getattr,
     .readlink   = sshfs_readlink,
@@ -848,6 +1161,7 @@ static struct fuse_operations sshfs_oper = {
     .truncate	= sshfs_truncate,
     .utime      = sshfs_utime,
     .open       = sshfs_open,
+    .release    = sshfs_release,
     .read       = sshfs_read,
     .write      = sshfs_write,
 };
@@ -858,29 +1172,33 @@ int main(int argc, char *argv[])
     int res;
     int argctr;
     char **newargv;
-    
 
     if (argc < 3) {
-        fprintf(stderr, "usage: %s [user@]host mountpoint [mount options]\n",
+        fprintf(stderr, "usage: %s [+][user@]host[:port] mountpoint [mount options]\n",
                 argv[0]);
         exit(1);
     }
 
     host = argv[1];
-    
-    res = start_ssh(host);
+    if (host[0] == '+')
+        res = connect_to(host+1);
+    else
+        res = start_ssh(host);
     if (res == -1)
         exit(1);
     
     res = sftp_init();
     if (res == -1)
         exit(1);
-    
+
+    res = processing_init();
+    if (res == -1)
+        exit(1);
+
     newargv = (char **) malloc((argc + 10) * sizeof(char *));
     newargv[0] = argv[0];
     for (argctr = 1; argctr < argc - 1; argctr++)
         newargv[argctr] = argv[argctr + 1];
-    newargv[argctr++] = "-s";
     newargv[argctr++] = "-omax_read=65536";
     newargv[argctr] = NULL;
     return fuse_main(argctr, newargv, &sshfs_oper);