TEMP: Wait toms patch

[Samba/vfs_proxy.git] / source4 / ntvfs / proxy / vfs_proxy.c

/*
   Unix SMB/PROXY implementation.

   CIFS PROXY NTVFS filesystem backend

   Copyright (C) Andrew Tridgell 2003
   Copyright (C) James J Myers 2003 <myersjj@samba.org>
   Copyright (C) Sam Liddicott <sam@liddicott.com>

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 3 of the License, or
   (at your option) any later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/
/*
  this implements a CIFS->CIFS NTVFS filesystem caching proxy.
  TODO:
  New read-ahead
  Delete cache
  Share cache states between processes
  Update to latest samba
  limit dirmons etc
  mapi delegated creds
*/

#define TALLOC_ABORT(why) smb_panic(why)
#warning handle SMB_FLAGS2_COMPRESSED flag from client: http://msdn2.microsoft.com/en-us/library/cc246254.aspx
#define __LOCATION__ (talloc_asprintf(debug_ctx(),"%s:%d %s",__FILE__,__LINE__,__FUNCTION__))
#define PROXY_NTIOCTL_MAXDATA 0x2000000

#include "includes.h"
#include "libcli/raw/libcliraw.h"
#include "libcli/smb_composite/smb_composite.h"
#include "auth/auth.h"
#include "auth/credentials/credentials.h"
#include "ntvfs/ntvfs.h"
#include "../lib/util/dlinklist.h"
#include "param/param.h"
#include "libcli/resolve/resolve.h"
#include "libcli/libcli.h"
#include "libcli/raw/ioctl.h"
#include "librpc/gen_ndr/ndr_misc.h"
#include "librpc/gen_ndr/ndr_proxy.h"
#include "librpc/ndr/ndr_table.h"
#include "lib/cache/cache.h"
#include "lib/compression/zlib.h"
#include "libcli/raw/raw_proto.h"
#include "librpc/gen_ndr/proxy.h"
#include "smb_server/smb_server.h"

#define fstrcmp(a,b) strcasecmp((a),(b))
#define fstrncmp(a,b,len) strncasecmp((a),(b),(len))

#define LOAD_CACHE_FILE_DATA(dest, src)  do { \
        dest.create_time=src.create_time; \
        dest.access_time=src.access_time; \
        dest.write_time=src.write_time; \
        dest.change_time=src.change_time; \
        dest.attrib=src.attrib; \
        dest.alloc_size=src.alloc_size; \
        dest.size=src.size; \
        dest.file_type=src.file_type; \
        dest.ipc_state=src.ipc_state; \
        dest.is_directory=src.is_directory; \
        dest.delete_pending=0; \
} while(0)

/* taken from #include "librpc/gen_ndr/proxy.h" */
struct proxy_file_info_data {
        /* first three are from ntcreatex */
        uint16_t file_type;
        uint16_t ipc_state;
        uint8_t  is_directory;
        NTSTATUS status_RAW_FILEINFO_BASIC_INFORMATION;
        uint32_t attrib;              /* RAW_FILEINFO_ALL_INFO | RAW_FILEINFO_BASIC_INFORMATION */
        NTTIME create_time;           /* RAW_FILEINFO_ALL_INFO | RAW_FILEINFO_BASIC_INFORMATION */
        NTTIME access_time;           /* RAW_FILEINFO_ALL_INFO | RAW_FILEINFO_BASIC_INFORMATION */
        NTTIME write_time;            /* RAW_FILEINFO_ALL_INFO | RAW_FILEINFO_BASIC_INFORMATION */
        NTTIME change_time;           /* RAW_FILEINFO_ALL_INFO | RAW_FILEINFO_BASIC_INFORMATION */
        NTSTATUS status_RAW_FILEINFO_ALL_INFO;
        uint32_t ea_size;             /* RAW_FILEINFO_ALL_INFO */
        uint64_t alloc_size;          /* RAW_FILEINFO_ALL_INFO */
        uint64_t size;                /* RAW_FILEINFO_ALL_INFO */
        uint32_t nlink;               /* RAW_FILEINFO_ALL_INFO */
        struct sws fname;             /* RAW_FILEINFO_ALL_INFO */
        uint8_t delete_pending;       /* RAW_FILEINFO_ALL_INFO */
        uint8_t directory;            /* RAW_FILEINFO_ALL_INFO */
        NTSTATUS status_RAW_FILEINFO_COMPRESSION_INFO;
        uint64_t compressed_size;     /* RAW_FILEINFO_COMPRESSION_INFO */
        uint16_t format;              /* RAW_FILEINFO_COMPRESSION_INFO */
        uint8_t unit_shift;           /* RAW_FILEINFO_COMPRESSION_INFO */
        uint8_t chunk_shift;          /* RAW_FILEINFO_COMPRESSION_INFO */
        uint8_t cluster_shift;        /* RAW_FILEINFO_COMPRESSION_INFO */
        NTSTATUS status_RAW_FILEINFO_INTERNAL_INFORMATION;
        uint64_t file_id;             /* RAW_FILEINFO_INTERNAL_INFORMATION */
        NTSTATUS status_RAW_FILEINFO_ACCESS_INFORMATION;
        uint32_t access_flags;        /* RAW_FILEINFO_ACCESS_INFORMATION */
        NTSTATUS status_RAW_FILEINFO_POSITION_INFORMATION;
        uint64_t position;            /* RAW_FILEINFO_POSITION_INFORMATION */
        NTSTATUS status_RAW_FILEINFO_MODE_INFORMATION;
        uint32_t mode;                /* RAW_FILEINFO_MODE_INFORMATION */
        NTSTATUS status_RAW_FILEINFO_ALIGNMENT_INFORMATION;
        uint32_t alignment_requirement; /* RAW_FILEINFO_ALIGNMENT_INFORMATION */
        NTSTATUS status_RAW_FILEINFO_ATTRIBUTE_TAG_INFORMATION;
        uint32_t reparse_tag;         /* RAW_FILEINFO_ATTRIBUTE_TAG_INFORMATION */
        uint32_t reparse_attrib;      /* RAW_FILEINFO_ATTRIBUTE_TAG_INFORMATION */
        NTSTATUS status_RAW_FILEINFO_STREAM_INFO;
        uint32_t num_streams;         /* RAW_FILEINFO_STREAM_INFO */
        struct info_stream *streams;         /* RAW_FILEINFO_STREAM_INFO */
};

#define valid_RAW_FILEINFO_BASIC_INFORMATION 1
#define valid_RAW_FILEINFO_ALL_INFO 2
#define valid_RAW_FILEINFO_COMPRESSION_INFO 3
#define valid_RAW_FILEINFO_INTERNAL_INFORMATION 4
#define valid_RAW_FILEINFO_STANDARD_INFO 8
#define valid_RAW_FILEINFO_ACCESS_INFORMATION 16
#define valid_RAW_FILEINFO_POSITION_INFORMATION 32
#define valid_RAW_FILEINFO_MODE_INFORMATION 64
#define valid_RAW_FILEINFO_ALIGNMENT_INFORMATION 128
#define valid_RAW_FILEINFO_ATTRIBUTE_TAG_INFORMATION 256
#define valid_RAW_FILEINFO_STREAM_INFO 512

struct file_metadata {
        int count;
        int valid;
        struct proxy_file_info_data info_data;
};

struct proxy_file {
        struct proxy_file *prev, *next;
        struct proxy_private* proxy;
        uint16_t fnum;
        struct ntvfs_handle *h;
        struct cache_file_entry *cache;
        /* filename might not be a char*, but if so, _size includes null */
        void* filename;
        int filename_size;
        int readahead_pending;
        /* *_OPLOCK_RETURN values */
        int oplock;
        /* read-only, shareable normal file open, can be cloned by similar opens */
        bool can_clone;
        /* If we have an oplock, then the file is NOT bigger than size, which lets
           us optimize reads */
        struct file_metadata *metadata;
};

struct proxy_private;

struct search_handle {
        struct search_handle *prev, *next;
        struct proxy_private *proxy;
        struct ntvfs_handle *h;
        uint16_t handle;
        union {
                struct smb_search_id id;
                uint32_t resume_key;
        } resume_index;
        struct search_cache_item *resume_item;
        enum smb_search_level level;
        enum smb_search_data_level data_level;
        /* search cache (if any) being used */
        struct search_cache *cache;
};

struct search_cache_item {
        struct search_cache_item *prev, *next;
        enum smb_search_data_level data_level;
        struct cache_file_entry *cache;
        union smb_search_data *file;
        struct file_metadata *metadata;
};
enum search_cache_status {
        SEARCH_CACHE_INCOMPLETE,
        SEARCH_CACHE_COMPLETE,
        SEARCH_CACHE_DEAD
};

struct fdirmon;
typedef void*(fdirmon_callback_fn)(void* data, struct fdirmon* fdirmon);
//NTSTATUS (*fn)(struct async_info*, void*, void*, NTSTATUS)

struct fdirmon {
        struct fdirmon *prev, *next;
        struct search_cache_item *items;

        struct proxy_private *proxy;

        union smb_notify *notify_io;
        struct smbcli_request *notify_req;
        uint16_t dir_fnum;
        char* dir;
        struct fdirmon_callback {
                struct fdirmon_callback *prev, *next;
                fdirmon_callback_fn *fn;
                void* data;
        } *callbacks;
};

struct search_cache {
        struct search_cache *prev, *next;
        struct search_cache_item *items;

        struct proxy_private *proxy;
        enum search_cache_status status;

        union smb_notify *notify_io;
        struct smbcli_request *notify_req;
        uint16_t dir_fnum;
        char* dir;

        struct search_cache_key {
                enum smb_search_level level;
                enum smb_search_data_level data_level;
                uint16_t search_attrib;
                const char *pattern;
                /* these only for trans2 */
                uint16_t flags;
                uint32_t storage_type;
        } key;
};
struct search_state {
        struct search_handle *search_handle;
        void* private;
        smbcli_search_callback callback;
        struct search_cache_item *last_item;
        uint16_t count;         /* count how many client receives */
        uint16_t all_count;     /* count how many we receive */
};

struct fs_attribute_info {
        uint32_t fs_attr;
        uint32_t max_file_component_length;
        struct smb_wire_string fs_type;
};

/* this is stored in ntvfs_private */
struct proxy_private {
        struct smbcli_tree *tree;
        struct smbcli_transport *transport;
        struct ntvfs_module_context *ntvfs;
        struct async_info *pending;
        struct proxy_file *files;
        struct proxy_file *closed_files;
        struct fdirmon *dirmons;
        struct search_cache *search_caches; /* cache's of find-first data */
        struct search_handle *search_handles; /* cache's of find-first data */
        bool map_generic;
        bool map_trans2;
        bool cache_enabled;
        int cache_readahead; /* default read-ahead window size */
        int cache_readaheadblock; /* size of each read-ahead request */
        ssize_t cache_validatesize; /* chunk size to validate, results in a read this size on remote server */
        char *remote_server;
        char *remote_share;
        struct cache_context *cache;
        struct fs_attribute_info *fs_attribute_info;
        int readahead_spare; /* amount of pending non-user generated requests */
        bool fake_oplock; /* useful for testing, smbclient never asks for oplock */
        bool fake_valid; /* useful for testing, smbclient never asks for oplock */
        uint16_t nttrans_fnum; /* we need a handle for non-proxy operations */
        bool enabled_cache_info;
        bool enabled_proxy_search;
        bool enabled_open_clone;
        bool enabled_extra_protocol;
        bool enabled_qpathinfo;
};

struct async_info_map;

/* a structure used to pass information to an async handler */
struct async_info {
        struct async_info *next, *prev;
        struct proxy_private *proxy;
        struct ntvfs_request *req;
        struct smbcli_request *c_req;
        struct proxy_file *f;
        struct async_info_map *chain;
        void *parms;
};

/* used to chain async callbacks */
struct async_info_map {
        struct async_info_map *next, *prev;
        NTSTATUS (*fn)(struct async_info*, void*, void*, NTSTATUS);
        void *parms1;
        void *parms2;
        struct async_info *async;
};

struct ntioctl_rpc_unmap_info {
        void* io;
        const struct ndr_interface_call *calls;
        const struct ndr_interface_table *table;
        uint32_t opnum;
};

/* a structure used to pass information to an async handler */
struct async_rpclite_send {
        const struct ndr_interface_call* call;
        void* struct_ptr;
};

#define SETUP_PID private->tree->session->pid = req->smbpid

#define RPCLITE_SETUP_FILE_HERE(f, h) do { \
        RPCLITE_SETUP_THIS_FILE_HERE(r->in.fnum, f, h); \
} while (0)

#define RPCLITE_SETUP_THIS_FILE_HERE(FNUM, f, h) do { \
        if ((h = ntvfs_find_handle(private->ntvfs, req, FNUM)) && \
            (f = ntvfs_handle_get_backend_data(h, ntvfs))) { \
            FNUM = f->fnum; \
        } else { \
            r->out.result = NT_STATUS_INVALID_HANDLE; \
            return NT_STATUS_OK; \
        } \
} while (0)

#define SETUP_FILE_HERE(f) do { \
        f = ntvfs_handle_get_backend_data(io->generic.in.file.ntvfs, ntvfs); \
        if (!f) return NT_STATUS_INVALID_HANDLE; \
        io->generic.in.file.fnum = f->fnum; \
} while (0)

#define SETUP_FILE do { \
        struct proxy_file *f; \
        SETUP_FILE_HERE(f); \
} while (0)

#define SETUP_PID_AND_FILE do { \
        SETUP_PID; \
        SETUP_FILE; \
} while (0)

/* remove the MAY_ASYNC from a request, useful for testing */
#define MAKE_SYNC_REQ(req) do { req->async_states->state &= ~NTVFS_ASYNC_STATE_MAY_ASYNC; } while(0)

#define PROXY_SERVER            "proxy:server"
#define PROXY_USER              "proxy:user"
#define PROXY_PASSWORD          "proxy:password"
#define PROXY_DOMAIN            "proxy:domain"
#define PROXY_SHARE             "proxy:share"
#define PROXY_USE_MACHINE_ACCT  "proxy:use-machine-account"
#define PROXY_MAP_GENERIC       "proxy:map-generic"
#define PROXY_MAP_TRANS2                "proxy:map-trans2"

#define PROXY_CACHE_ENABLED             "proxy:cache-enabled"
#define PROXY_CACHE_ENABLED_DEFAULT             false

#define PROXY_CACHE_READAHEAD                  "proxy:cache-readahead"
#define PROXY_CACHE_READAHEAD_DEFAULT          32768
/* size of each read-ahead request. */
#define PROXY_CACHE_READAHEAD_BLOCK            "proxy:cache-readaheadblock"
/* the read-ahead block should always be less than max negotiated data */
#define PROXY_CACHE_READAHEAD_BLOCK_DEFAULT    4096

#define PROXY_CACHE_VALIDATE_SIZE       "proxy:validate-size"
#define PROXY_CACHE_VALIDATE_SIZE_DEFAULT       256 /* 10M */

#define PROXY_FAKE_OPLOCK       "proxy:fake-oplock"
#define PROXY_FAKE_OPLOCK_DEFAULT               false

#define PROXY_FAKE_VALID        "proxy:fake-valid"
#define PROXY_FAKE_VALID_DEFAULT                false

/* how many read-ahead requests can be pending per mid */
#define PROXY_REQUEST_LIMIT     "proxy:request-limit"
#define PROXY_REQUEST_LIMIT_DEFAULT     100

#define PROXY_USE_MACHINE_ACCT_DEFAULT  false
/* These two really should be: true, and possibly not even configurable */
#define PROXY_MAP_GENERIC_DEFAULT               true
#define PROXY_MAP_TRANS2_DEFAULT                true

/* is the remote server a proxy? */
#define PROXY_REMOTE_SERVER(private) \
        ((private)->tree->session->transport->negotiate.capabilities & CAP_COMPRESSION \
         && (strcmp("A:",private->tree->device)==0) \
         && (private->nttrans_fnum!=0) \
         && (private->enabled_extra_protocol))

/* A few forward declarations */
static NTSTATUS sync_chain_handler(struct smbcli_request *c_req);
static void async_chain_handler(struct smbcli_request *c_req);
static void async_read_handler(struct smbcli_request *c_req);
static NTSTATUS proxy_rpclite(struct ntvfs_module_context *ntvfs,
                           struct ntvfs_request *req, union smb_ioctl *io);

struct smbcli_request *smbcli_ndr_request_ntioctl_send(
                                                struct smbcli_tree *tree,
                                                struct ntvfs_module_context *ntvfs,
                                                const struct ndr_interface_table *table,
                                                uint32_t opnum, void *r);
struct smbcli_request *proxy_smb_raw_read_send(struct ntvfs_module_context *ntvfs,
                                                union smb_read *io, struct proxy_file *f, struct proxy_Read *r);
NTSTATUS proxy_smb_raw_read(struct ntvfs_module_context *ntvfs,
                                                union smb_read *io, struct proxy_file *f);
struct smbcli_request *proxy_smb_raw_write_send(struct ntvfs_module_context *ntvfs,
                                                union smb_write *io, struct proxy_file *f);
NTSTATUS proxy_smb_raw_write(struct ntvfs_module_context *ntvfs,
                                                 union smb_write *io, struct proxy_file *f);
static NTSTATUS async_read_fragment(struct async_info *async, void* io1, void* io2, NTSTATUS status);

struct smb_wire_string talloc_smb_wire_string_dup(void* mem_ctx, const struct smb_wire_string* string)
{
        struct smb_wire_string result;
        result.private_length=string->private_length;
        result.s=talloc_strndup(mem_ctx, string->s, string->private_length);
        DEBUG(5,("%s: %s\n",__FUNCTION__, string->s));
        return result;
}

#define sws_dup(mem_ctx, dest, src) (\
        dest=talloc_smb_wire_string_dup(NULL, &(src)), \
        (dest.s==NULL && src.s!=NULL))

/* These needs replacing with something more canonical perhaps */
static char* talloc_dirname(void* mem_ctx, const char* path) {
        const char* dir;

        if ((dir=strrchr(path,'\\'))) {
                return talloc_strndup(mem_ctx, path, (dir - path));
        } else {
                return talloc_strdup(mem_ctx,"");
        }
}

/*
  a handler for oplock break events from the server - these need to be passed
  along to the client
 */
static bool oplock_handler(struct smbcli_transport *transport, uint16_t tid, uint16_t fnum, uint8_t level, void *p_private)
{
        struct proxy_private *private = p_private;
        NTSTATUS status;
        struct ntvfs_handle *h = NULL;
        struct proxy_file *f;
        bool result=true;

        /* because we clone handles, there may be more than one match */
        for (f=private->files; f; f=f->next) {
                if (f->fnum != fnum) continue;
                h = f->h;

                if (level==OPLOCK_BREAK_TO_LEVEL_II) {
                        f->oplock=LEVEL_II_OPLOCK_RETURN;
                } else {
                        /* If we don't have an oplock, then we can't rely on the cache */
                        cache_handle_stale(f);
                        f->oplock=NO_OPLOCK_RETURN;
                }

                DEBUG(5,("vfs_proxy: sending oplock break level %d for fnum %d\n", level, fnum));
                status = ntvfs_send_oplock_break(private->ntvfs, h, level);
                if (!NT_STATUS_IS_OK(status)) result=false;
        }
        if (!h) {
                DEBUG(5,("vfs_proxy: ignoring oplock break level %d for fnum %d\n", level, fnum));
        }
        return result;
}

/*
  get file handle from clients fnum, (from ntvfs/ipc/vfs_ipc.c at metze suggestion)
*/
static struct ntvfs_handle *ntvfs_find_handle(struct ntvfs_module_context *ntvfs,
                                              struct ntvfs_request *req,
                                              uint16_t fnum)
{
        DATA_BLOB key;
        uint16_t _fnum;

        /*
         * the fnum is already in host byteorder
         * but ntvfs_handle_search_by_wire_key() expects
         * network byteorder
         */
        SSVAL(&_fnum, 0, fnum);
        key = data_blob_const(&_fnum, 2);

        return ntvfs_handle_search_by_wire_key(ntvfs, req, &key);
}

/*
  connect to a share - used when a tree_connect operation comes in.
*/
static NTSTATUS proxy_connect(struct ntvfs_module_context *ntvfs,
                             struct ntvfs_request *req, const char *sharename)
{
        NTSTATUS status;
        struct proxy_private *private;
        const char *host, *user, *pass, *domain, *remote_share;
        struct smb_composite_connect io;
        struct composite_context *creq;
        struct share_config *scfg = ntvfs->ctx->config;
        int nttrans_fnum;

        struct cli_credentials *credentials;
        bool machine_account;

        /* Here we need to determine which server to connect to.
         * For now we use parametric options, type proxy.
         * Later we will use security=server and auth_server.c.
         */
        host = share_string_option(scfg, PROXY_SERVER, NULL);
        user = share_string_option(scfg, PROXY_USER, NULL);
        pass = share_string_option(scfg, PROXY_PASSWORD, NULL);
        domain = share_string_option(scfg, PROXY_DOMAIN, NULL);
        remote_share = share_string_option(scfg, PROXY_SHARE, NULL);
        if (!remote_share) {
                remote_share = sharename;
        }

        machine_account = share_bool_option(scfg, PROXY_USE_MACHINE_ACCT, PROXY_USE_MACHINE_ACCT_DEFAULT);

        private = talloc_zero(ntvfs, struct proxy_private);
        if (!private) {
                return NT_STATUS_NO_MEMORY;
        }

        ntvfs->private_data = private;

        if (!host) {
                DEBUG(1,("PROXY backend: You must supply server\n"));
                return NT_STATUS_INVALID_PARAMETER;
        }

        if (user && pass) {
                DEBUG(5, ("PROXY backend: Using specified password\n"));
                credentials = cli_credentials_init(private);
                if (!credentials) {
                        return NT_STATUS_NO_MEMORY;
                }
                cli_credentials_set_conf(credentials, ntvfs->ctx->lp_ctx);
                cli_credentials_set_username(credentials, user, CRED_SPECIFIED);
                if (domain) {
                        cli_credentials_set_domain(credentials, domain, CRED_SPECIFIED);
                }
                cli_credentials_set_password(credentials, pass, CRED_SPECIFIED);
        } else if (machine_account) {
                DEBUG(5, ("PROXY backend: Using machine account\n"));
                credentials = cli_credentials_init(private);
                cli_credentials_set_conf(credentials, ntvfs->ctx->lp_ctx);
                if (domain) {
                        cli_credentials_set_domain(credentials, domain, CRED_SPECIFIED);
                }
                status = cli_credentials_set_machine_account(credentials, ntvfs->ctx->lp_ctx);
                if (!NT_STATUS_IS_OK(status)) {
                        return status;
                }
        } else if (req->session_info->credentials) {
                DEBUG(5, ("PROXY backend: Using delegated credentials\n"));
                credentials = req->session_info->credentials;
        } else {
                DEBUG(1,("PROXY backend: NO delegated credentials found: You must supply server, user and password or the client must supply delegated credentials\n"));
                return NT_STATUS_INVALID_PARAMETER;
        }

        /* connect to the server, using the smbd event context */
        io.in.dest_host = host;
        io.in.dest_ports = lp_smb_ports(ntvfs->ctx->lp_ctx);
        io.in.socket_options = lp_socket_options(ntvfs->ctx->lp_ctx);
        io.in.called_name = host;
        io.in.credentials = credentials;
        io.in.fallback_to_anonymous = false;
        io.in.workgroup = lp_workgroup(ntvfs->ctx->lp_ctx);
        io.in.service = remote_share;
        io.in.service_type = "?????";
        io.in.iconv_convenience = lp_iconv_convenience(ntvfs->ctx->lp_ctx);
        io.in.gensec_settings = lp_gensec_settings(private, ntvfs->ctx->lp_ctx);
        lp_smbcli_options(ntvfs->ctx->lp_ctx, &io.in.options);
        lp_smbcli_session_options(ntvfs->ctx->lp_ctx, &io.in.session_options);

        creq = smb_composite_connect_send(&io, private,
                                          lp_resolve_context(ntvfs->ctx->lp_ctx),
                                          ntvfs->ctx->event_ctx);
        status = smb_composite_connect_recv(creq, private);
        NT_STATUS_NOT_OK_RETURN(status);

        private->tree = io.out.tree;

        private->transport = private->tree->session->transport;
        SETUP_PID;
        private->ntvfs = ntvfs;

        ntvfs->ctx->fs_type = talloc_strdup(ntvfs->ctx, io.out.tree->fs_type);
        NT_STATUS_HAVE_NO_MEMORY(ntvfs->ctx->fs_type);
        ntvfs->ctx->dev_type = talloc_strdup(ntvfs->ctx, io.out.tree->device);
        NT_STATUS_HAVE_NO_MEMORY(ntvfs->ctx->dev_type);

        /* we need to receive oplock break requests from the server */
        smbcli_oplock_handler(private->transport, oplock_handler, private);

        private->map_generic = share_bool_option(scfg, PROXY_MAP_GENERIC, PROXY_MAP_GENERIC_DEFAULT);

        private->map_trans2 = share_bool_option(scfg, PROXY_MAP_TRANS2, PROXY_MAP_TRANS2_DEFAULT);

        private->cache_validatesize = 1024 * (long long) share_int_option(scfg, PROXY_CACHE_VALIDATE_SIZE, PROXY_CACHE_VALIDATE_SIZE_DEFAULT);

        if (strcmp("A:",private->tree->device)==0) {
                private->cache_enabled = share_bool_option(scfg, PROXY_CACHE_ENABLED, PROXY_CACHE_ENABLED_DEFAULT);
                private->cache_readahead = share_int_option(scfg, PROXY_CACHE_READAHEAD, PROXY_CACHE_READAHEAD_DEFAULT);
                private->cache_readaheadblock = share_int_option(scfg, PROXY_CACHE_READAHEAD_BLOCK,
                                        MIN(private->cache_readahead,PROXY_CACHE_READAHEAD_BLOCK_DEFAULT));
                private->fake_oplock = share_bool_option(scfg, PROXY_FAKE_OPLOCK, PROXY_FAKE_OPLOCK_DEFAULT);
                private->fake_valid = share_bool_option(scfg, PROXY_FAKE_VALID, PROXY_FAKE_VALID_DEFAULT);
                private->readahead_spare = share_int_option(scfg, PROXY_REQUEST_LIMIT, PROXY_REQUEST_LIMIT_DEFAULT);
                private->cache = new_cache_context(private, lp_proxy_cache_root(ntvfs->ctx->lp_ctx), host, remote_share);
                private->enabled_cache_info=true;
                private->enabled_proxy_search=true;
                private->enabled_open_clone=true;
                private->enabled_extra_protocol=true;
                private->enabled_qpathinfo=true;

                DEBUG(0,("proxy tree connect caching for: %s (%s : %s) %s read-ahead: %d\n",
                                 remote_share, private->tree->device,private->tree->fs_type,
                             (private->cache_enabled)?"enabled":"disabled",
                              private->cache_readahead));
        } else {
                private->cache_enabled = false;
                DEBUG(0,("No caching or read-ahead for: %s (%s : %s)\n",
                                 remote_share, private->tree->device,private->tree->fs_type));
        }

        private->remote_server = strlower_talloc(private, host);
        private->remote_share = strlower_talloc(private, remote_share);

        /* some proxy operations will not be performed on files, so open a handle
           now that we can use for such things. We won't bother to close it on
           shutdown, as the remote server ought to be able to close it for us
           and we might be shutting down because the remote server went away and
           so we don't want to delay further */
        nttrans_fnum=smbcli_nt_create_full(private->tree, "\\",
                                                    0,
                                                    SEC_FILE_READ_DATA,
                                                    FILE_ATTRIBUTE_NORMAL,
                                                    NTCREATEX_SHARE_ACCESS_MASK,
                                                    NTCREATEX_DISP_OPEN,
                                                    NTCREATEX_OPTIONS_DIRECTORY,
                                                    NTCREATEX_IMPERSONATION_IMPERSONATION);
        if (nttrans_fnum < 0) {
                DEBUG(5,("Could not open handle for ntioctl %d\n",private->nttrans_fnum));
                //return NT_STATUS_UNSUCCESSFUL;
        }
        private->nttrans_fnum=nttrans_fnum;
        DEBUG(5,("Got nttrans handle %d\n",private->nttrans_fnum));

        return NT_STATUS_OK;
}

/*
  disconnect from a share
*/
static NTSTATUS proxy_disconnect(struct ntvfs_module_context *ntvfs)
{
        struct proxy_private *private = ntvfs->private_data;
        struct async_info *a, *an;
        struct search_cache *s;

        /* first clean up caches because they have a pending request that
           they will try and clean up later and fail during talloc_free */
        for (s=private->search_caches; s; s=s->next) {
                if (s->notify_req) {
                        talloc_unlink(s, s->notify_req);
                        s->notify_req=NULL;
                }
                s->dir_fnum=65535;
        }

        /* first cleanup pending requests */
        for (a=private->pending; a; a = an) {
                an = a->next;
                smbcli_request_destroy(a->c_req);
                talloc_free(a);
        }

        talloc_free(private);
        ntvfs->private_data = NULL;

        return NT_STATUS_OK;
}

/*
  destroy an async info structure
*/
static int async_info_destructor(struct async_info *async)
{
        DLIST_REMOVE(async->proxy->pending, async);
        return 0;
}

/*
  a handler for simple async replies
  this handler can only be used for functions that don't return any
  parameters (those that just return a status code)
 */
static void async_simple(struct smbcli_request *c_req)
{
        struct async_info *async = c_req->async.private;
        struct ntvfs_request *req = async->req;
        req->async_states->status = smbcli_request_simple_recv(c_req);
        talloc_free(async);
        req->async_states->send_fn(req);
}

/* hopefully this will optimize away */
#define TYPE_CHECK(type,check) do { \
        type=check; \
        t=t; \
} while (0)

/* save some typing for the simple functions */
#define ASYNC_RECV_TAIL_F_ORPHAN(io, async_fn, file, achain, error) do { \
        if (!c_req) return (error); \
        ASYNC_RECV_TAIL_F_ORPHAN_NE(io, async_fn, file, achain); \
        if (! c_req->async.private) return (error); \
} while(0)

#define ASYNC_RECV_TAIL_F_ORPHAN_NE(io, async_fn, file, achain) do { \
        TYPE_CHECK(void (*t)(struct smbcli_request *),async_fn); \
        { \
                struct async_info *async; \
                async = talloc(req, struct async_info); \
                if (async) { \
                        async->parms = io; \
                        async->req = req; \
                        async->f = file; \
                        async->proxy = private; \
                        async->c_req = c_req; \
                        async->chain = achain; \
                        DLIST_ADD(private->pending, async); \
                        c_req->async.private = async; \
                        talloc_set_destructor(async, async_info_destructor); \
                } \
        } \
        c_req->async.fn = async_fn; \
} while (0)

#define ASYNC_RECV_TAIL_F(io, async_fn, file) do { \
        if (!c_req) return NT_STATUS_UNSUCCESSFUL; \
        TYPE_CHECK(void (*t)(struct smbcli_request *),async_fn); \
        { \
                struct async_info *async; \
                async = talloc(req, struct async_info); \
                if (!async) return NT_STATUS_NO_MEMORY; \
                async->parms = io; \
                async->req = req; \
                async->f = file; \
                async->proxy = private; \
                async->c_req = c_req; \
                DLIST_ADD(private->pending, async); \
                c_req->async.private = async; \
                talloc_set_destructor(async, async_info_destructor); \
        } \
        c_req->async.fn = async_fn; \
        req->async_states->state |= NTVFS_ASYNC_STATE_ASYNC; \
        return NT_STATUS_OK; \
} while (0)

#define ASYNC_RECV_TAIL(io, async_fn) ASYNC_RECV_TAIL_F(io, async_fn, NULL)

#define SIMPLE_ASYNC_TAIL ASYNC_RECV_TAIL(NULL, async_simple)

/* managers for chained async-callback.
   The model of async handlers has changed.
   backend async functions should be of the form:
     NTSTATUS (*fn)(struct async_info*, void*, void*, NTSTATUS);
   And if async->c_req is NULL then an earlier chain has already rec'd the
   request.
   ADD_ASYNC_RECV_TAIL is used to add chained handlers.
   The chained handler manager async_chain_handler is installed the usual way
   and uses the io pointer to point to the first async_map record
   static void async_chain_handler(struct smbcli_request *c_req).
   It is safe to call ADD_ASYNC_RECV_TAIL before the chain manager is installed
   and often desirable.
 */
/* async_chain_handler has an async_info struct so that it can be safely inserted
   into pending, but the io struct will point to (struct async_info_map *)
   chained async_info_map will be in c_req->async.private */
#define ASYNC_RECV_TAIL_HANDLER_ORPHAN(io, async_fn) do { \
        if (c_req->async.fn) return (NT_STATUS_UNSUCCESSFUL); \
        ASYNC_RECV_TAIL_F_ORPHAN(io, async_fn, f, c_req->async.private, NT_STATUS_UNSUCCESSFUL); \
} while(0)

#define ASYNC_RECV_TAIL_HANDLER(io, async_fn) do { \
        if (c_req->async.fn) return (NT_STATUS_UNSUCCESSFUL); \
        ASYNC_RECV_TAIL_F_ORPHAN(io, async_fn, f, c_req->async.private, NT_STATUS_UNSUCCESSFUL); \
        req->async_states->state |= NTVFS_ASYNC_STATE_ASYNC; \
        return NT_STATUS_OK; \
} while(0)

/*
        DEBUG(0,("ADD_ASYNC_RECV_TAIL %s %s:%d\n\t%p=%s %p\n\t%p=%s %p\n\t%p=%s %p\n\t%p=%s %p\n\t%s\n", __FUNCTION__,__FILE__,__LINE__, \
                creq, creq?talloc_get_name(creq):NULL, creq?talloc_get_name(creq):NULL,\
                io1, io1?talloc_get_name(io1):NULL, io1?talloc_get_name(io1):NULL, \
                io2, io2?talloc_get_name(io2):NULL, io2?talloc_get_name(io2):NULL, \
                file, file?"file":"null", file?"file":"null", #async_fn)); \
*/
#define ADD_ASYNC_RECV_TAIL(creq, io1, io2, file, async_fn, error) do { \
        if (! creq) { \
                DEBUG(5,("%s: ADD_ASYNC_RECV_TAIL no creq\n",__FUNCTION__)); \
                return (error); \
        } else { \
                struct async_info_map *async_map=talloc(NULL, struct async_info_map); \
                if (! async_map) { \
                        DEBUG(5,("%s: ADD_ASYNC_RECV_TAIL no async_map\n",__FUNCTION__)); \
                        return (error); \
                } \
                async_map->async=talloc(async_map, struct async_info); \
                if (! async_map->async) { \
                        DEBUG(5,("%s: ADD_ASYNC_RECV_TAIL no async_map->async\n",__FUNCTION__)); \
                        return (error); \
                } \
                async_map->parms1=io1; \
                async_map->parms2=io2; \
                async_map->fn=async_fn; \
                async_map->async->parms = io1; \
                async_map->async->req = req; \
                async_map->async->f = file; \
                async_map->async->proxy = private; \
                async_map->async->c_req = creq; \
                /* If async_chain_handler is installed, get the list from param */ \
                if (creq->async.fn == async_chain_handler || creq->async.fn == async_read_handler) { \
                        struct async_info *i=creq->async.private; \
                        DLIST_ADD_END(i->chain, async_map, struct async_info_map *); \
                } else if (creq->async.fn) { \
                        /* incompatible handler installed */ \
                        DEBUG(5,("%s: ADD_ASYNC_RECV_TAIL incompatible handler already installed\n",__FUNCTION__)); \
                        return (error); \
                } else { \
                        DLIST_ADD_END(creq->async.private, async_map, struct async_info_map *); \
                } \
        } \
} while(0)

static void async_dirmon_notify(struct smbcli_request *c_req)
{
        struct async_info *async = c_req->async.private;
        struct ntvfs_request *req = async->req;
        struct fdirmon *dirmon;
        struct fdirmon_callback *callback;
        struct proxy_private *proxy = async->proxy;
        int f;

        NTSTATUS status;

        dirmon = talloc_get_type_abort((void*)async->f, struct fdirmon);
        DEBUG(5,("%s: dirmon %s invalidated\n",__LOCATION__, dirmon->dir));

        status = smb_raw_changenotify_recv(c_req, req, async->parms);
        DEBUG(5,("%s: update status %s\n",__LOCATION__, get_friendly_nt_error_msg (status)));

        if (dirmon->notify_req) {
                talloc_unlink(dirmon, dirmon->notify_req);
                dirmon->notify_req=NULL;
        }
        /* Mark closed cached files as invalid if they changed, as they will be
           assuming cache is valid if a dirmon exists and hasn't invalidated it */
        for(f=0; f<dirmon->notify_io->nttrans.out.num_changes; f++) {
                DEBUG(1,("DIRMON: %s changed\n",dirmon->notify_io->nttrans.out.changes[f].name.s));
        }
        DLIST_FOR_EACH(dirmon->callbacks, callback, callback->fn(callback->data, dirmon));
        /* So nothing can find it even if there are still in-use references */
        DLIST_REMOVE(proxy->dirmons, dirmon);
        if (dirmon->dir_fnum!=65535) {
                struct smbcli_request *req;
                union smb_close close_parms;
                close_parms.close.level = RAW_CLOSE_CLOSE;
                close_parms.close.in.file.fnum = dirmon->dir_fnum;
                close_parms.close.in.write_time = 0;

                /* destructor may be called from a notify response and won't be able
                   to wait on this close response, not that we care anyway */
                req=smb_raw_close_send(proxy->tree, &close_parms);

                DEBUG(5,("%s: Close dir_fnum: %d %p\n",__LOCATION__, dirmon->dir_fnum, req));
                dirmon->dir_fnum=65535;
        }
        talloc_free(async);
        talloc_free(dirmon);
}

struct fdirmon* get_fdirmon(struct proxy_private *proxy, const char* path, bool dir_only) {
        const char *file;
        int pathlen;

        if ((file=strrchr(path,'\\'))) {
                if (dir_only) {
                        pathlen = file - path;
                        file++;
                } else {
                        pathlen=strlen(path);
                }
        } else {
                file = path;
                pathlen = 0;
        }

        struct fdirmon *dirmon;
        /* see if we have a matching dirmon */
        DLIST_FIND(proxy->dirmons, dirmon, (strlen(dirmon->dir) == pathlen && fstrncmp(path, dirmon->dir, pathlen)==0));
        if (! dirmon) {
                int saved_timeout;

                DEBUG(5,("%s: allocating new dirmon for %s\n",__FUNCTION__,path));
                dirmon=talloc_zero(proxy, struct fdirmon);
                if (! dirmon) {
                        goto error;
                }
                if (! (dirmon->dir=talloc_strndup(dirmon, path, pathlen))) {
                        goto error;
                }
                if (! (dirmon->notify_io=talloc_zero(dirmon, union smb_notify))) {
                        goto error;
                }

                dirmon->dir_fnum=smbcli_nt_create_full(proxy->tree, dirmon->dir,
                                                    0,
                                                    SEC_FILE_READ_DATA,
                                                    FILE_ATTRIBUTE_NORMAL,
                                                    NTCREATEX_SHARE_ACCESS_MASK,
                                                    NTCREATEX_DISP_OPEN,
                                                    NTCREATEX_OPTIONS_DIRECTORY,
                                                    NTCREATEX_IMPERSONATION_IMPERSONATION);

                if (dirmon->dir_fnum==65535) {
                        DEBUG(5,("%s: smbcli_nt_create_full %s failed\n",__FUNCTION__, dirmon->dir));
                        goto error;
                }

                saved_timeout = proxy->transport->options.request_timeout;
                /* request notify changes on cache before we start to fill it */
                dirmon->notify_io->nttrans.level=RAW_NOTIFY_NTTRANS;
                dirmon->notify_io->nttrans.in.completion_filter=FILE_NOTIFY_CHANGE_ANY;
                dirmon->notify_io->nttrans.in.file.fnum=dirmon->dir_fnum;
                dirmon->notify_io->nttrans.in.recursive=false;
                dirmon->notify_io->nttrans.in.buffer_size=10240;
                proxy->transport->options.request_timeout = 0;
                dirmon->notify_req=smb_raw_changenotify_send(proxy->tree, dirmon->notify_io);
                /* Make the request hang around so we can tell if it needs cancelling */
                proxy->transport->options.request_timeout = saved_timeout;

                if (! dirmon->notify_req) {
                        goto error;
                }else {
                        struct ntvfs_request *req=NULL;
                        struct smbcli_request *c_req=dirmon->notify_req;
                        union smb_notify *io=dirmon->notify_io;
                        struct proxy_private *private=proxy;

                        talloc_reference(dirmon, dirmon->notify_req);
                        ASYNC_RECV_TAIL_F_ORPHAN_NE(io, async_dirmon_notify,
                                                 (void*) dirmon, c_req->async.private);
                        DLIST_ADD(private->dirmons, dirmon);
                }
        }

        return dirmon;
    error:
        DEBUG(3,("%s: failed to allocate dirmon\n",__FUNCTION__));
        talloc_free(dirmon);
        return NULL;
}

bool dirmon_add_callback(struct fdirmon *dirmon, fdirmon_callback_fn *fn, void* data) {
        struct fdirmon_callback *callback=talloc_zero(dirmon, struct fdirmon_callback);
        if (! callback) {
                return false;
        }
        callback->data=data;
        callback->fn=fn;
        DLIST_ADD(dirmon->callbacks, callback);
        return true;
}

/* try and unify cache open function interface with this macro */
#define cache_open(cache_context, f, io, oplock, readahead_window) \
        (io->generic.level == RAW_OPEN_NTCREATEX && \
         io->generic.in.create_options & NTCREATEX_OPTIONS_OPEN_BY_FILE_ID)\
         ?(cache_fileid_open(cache_context, f, (const uint64_t*)(io->generic.in.fname), oplock, readahead_window))\
         :(cache_filename_open(cache_context, f, SMB_OPEN_IN_FILE(io), oplock, readahead_window))

struct search_cache* find_partial_search_cache(struct search_cache* search_cache, const struct search_cache_key* search_cache_key) {
        struct search_cache* result;
        DLIST_FIND(search_cache, result,
                   (result->key.level == search_cache_key->level) &&
                   (result->key.data_level == search_cache_key->data_level) &&
                   (result->key.search_attrib == search_cache_key->search_attrib) &&
                   (result->key.flags == search_cache_key->flags) &&
                   (result->key.storage_type == search_cache_key->storage_type) &&
                   (fstrcmp(result->key.pattern, search_cache_key->pattern) == 0));
        DEBUG(5,("%s: found %p\n",__LOCATION__,result));
        return result;
}
struct search_cache* find_search_cache(struct search_cache* search_cache, const struct search_cache_key* search_cache_key) {
        struct search_cache* result = find_partial_search_cache(search_cache, search_cache_key);
        if (result && result->status == SEARCH_CACHE_COMPLETE) {
                DEBUG(5,("%s: found complete %p\n",__LOCATION__,result));
                return result;
        }
        DEBUG(5,("%s: found INCOMPLETE %p\n",__LOCATION__,result));
        return NULL;
}

uint16_t smbsrv_fnum(struct ntvfs_handle *h) {
        uint16_t fnum;
        smbsrv_push_fnum((uint8_t *)&fnum, 0, h);
        return SVAL(&fnum, 0);
}

static void async_search_cache_notify(struct smbcli_request *c_req)
{
        struct async_info *async = c_req->async.private;
        struct ntvfs_request *req = async->req;
        struct search_cache *search_cache;
        NTSTATUS status;

        DEBUG(5,("%s: search cache %p invalidated\n",__LOCATION__, (void*)async->f));

        search_cache = talloc_get_type_abort((void*)async->f, struct search_cache);

        status = smb_raw_changenotify_recv(c_req, req, async->parms);

        DEBUG(5,("%s: update status is %s\n",__LOCATION__, get_friendly_nt_error_msg (status)));

        search_cache->notify_req=NULL;
        /* dispose of the search_cache */
        search_cache->status=SEARCH_CACHE_DEAD;
        /* So nothing can find it even if there are still in-use references */
        DLIST_REMOVE(search_cache->proxy->search_caches, search_cache);
        /* free it */
        //talloc_steal(async, search_cache);
        //talloc_unlink(async->proxy, search_cache);
        if (search_cache->dir_fnum!=65535) {
                struct smbcli_request *req;
                union smb_close close_parms;
                close_parms.close.level = RAW_CLOSE_CLOSE;
                close_parms.close.in.file.fnum = search_cache->dir_fnum;
                close_parms.close.in.write_time = 0;

                /* destructor may be called from a notify response and won't be able
                   to wait on this close response, not that we care anyway */
                req=smb_raw_close_send(search_cache->proxy->tree, &close_parms);

                DEBUG(5,("%s: Close dir_fnum: %d %p\n",__LOCATION__, search_cache->dir_fnum, req));
                search_cache->dir_fnum=65535;
        }
        talloc_free(async);
}

/*
  destroy a search handle
*/
static int search_handle_destructor(struct search_handle *s)
{
        DLIST_REMOVE(s->proxy->search_handles, s);
        DEBUG(5,("%s: handle destructor %p\n",__LOCATION__,s));
        return 0;
}
static int search_cache_destructor(struct search_cache *s)
{
        NTSTATUS status;

        DLIST_REMOVE(s->proxy->search_caches, s);
        DEBUG(5,("%s: cache destructor %p\n",__LOCATION__,s));
        if (s->notify_req) {
                status=smb_raw_ntcancel(s->notify_req);
                s->notify_req=NULL;
                DEBUG(5,("%s: Cancel notification %s\n",__LOCATION__,get_friendly_nt_error_msg (status)));
        }
        if (s->dir_fnum!=65535) {
                struct smbcli_request *req;
                union smb_close close_parms;
                close_parms.close.level = RAW_CLOSE_CLOSE;
                close_parms.close.in.file.fnum = s->dir_fnum;
                close_parms.close.in.write_time = 0;

                /* destructor may be called from a notify response and won't be able
                   to wait on this close response, not that we care anyway */
                req=smb_raw_close_send(s->proxy->tree, &close_parms);

                DEBUG(5,("%s: Close dir_fnum: %d %p\n",__LOCATION__, s->dir_fnum, req));
                s->dir_fnum=65535;
        }
        return 0;
}

struct search_cache* new_search_cache(struct proxy_private *private, struct search_cache_key* key) {
        /* need to opendir the folder being searched so we can get a notification */
        uint16_t dir_fnum=65535;
        struct search_cache *search_cache=NULL;

        search_cache=talloc_zero(private, struct search_cache);
        DEBUG(5,("%s: Start new cache %p for %s\n",__LOCATION__, search_cache, key->pattern));
        if (! search_cache) {
                return NULL;
        }
        search_cache->proxy=private;
        if (! (search_cache->dir=talloc_dirname(search_cache, key->pattern))) {
                goto error;
        }
        if (! (search_cache->notify_io=talloc_zero(search_cache, union smb_notify))) {
                goto error;
        }
        search_cache->key=*key;
        /* make private copy of pattern now that we need it AND have something to own it */
        if (! (search_cache->key.pattern=talloc_strdup(search_cache, search_cache->key.pattern))) {
                goto error;
        }
        dir_fnum=smbcli_nt_create_full(private->tree, search_cache->dir,
                                                    0,
                                                    SEC_FILE_READ_DATA,
                                                    FILE_ATTRIBUTE_NORMAL,
                                                    NTCREATEX_SHARE_ACCESS_MASK,
                                                    NTCREATEX_DISP_OPEN,
                                                    NTCREATEX_OPTIONS_DIRECTORY,
                                                    NTCREATEX_IMPERSONATION_IMPERSONATION);
        DEBUG(5,("%s: %d=opendir on %s\n",__LOCATION__,dir_fnum, search_cache->dir));
        if (dir_fnum==65535) {
                goto error;
        }
        /* The destructor will close the handle */
        talloc_set_destructor(search_cache, search_cache_destructor);
        search_cache->dir_fnum=dir_fnum;
        DEBUG(5,("%s: Start new cache %p, dir_fnum %d\n",__LOCATION__, search_cache, dir_fnum));

        {
                int saved_timeout = private->transport->options.request_timeout;

                /* request notify changes on cache before we start to fill it */
                search_cache->notify_io->nttrans.level=RAW_NOTIFY_NTTRANS;
                search_cache->notify_io->nttrans.in.completion_filter=FILE_NOTIFY_CHANGE_ANY;
                search_cache->notify_io->nttrans.in.file.fnum=dir_fnum;
                search_cache->notify_io->nttrans.in.recursive=false;
                search_cache->notify_io->nttrans.in.buffer_size=1024;
                private->transport->options.request_timeout = 0;
                search_cache->notify_req=smb_raw_changenotify_send(private->tree, search_cache->notify_io);
                /* Make the request hang around so we can tell if it needs cancelling */
                private->transport->options.request_timeout = saved_timeout;

                if (! search_cache->notify_req) {
                        goto error;
                } else {
                        struct ntvfs_request *req=NULL;
                        struct smbcli_request *c_req=search_cache->notify_req;
                        union smb_notify *io=search_cache->notify_io;

                        talloc_reference(search_cache, search_cache->notify_req);
                        ASYNC_RECV_TAIL_F_ORPHAN_NE(io, async_search_cache_notify,
                                                 (void*) search_cache, c_req->async.private);
                        DLIST_ADD_END(private->search_caches, search_cache, struct search_cache*);
                }
        }
        return search_cache;
    error:
        talloc_free(search_cache);
        return NULL;
}

/*
  delete a file - the dirtype specifies the file types to include in the search.
  The name can contain PROXY wildcards, but rarely does (except with OS/2 clients)
*/
static NTSTATUS proxy_unlink(struct ntvfs_module_context *ntvfs,
                            struct ntvfs_request *req, union smb_unlink *unl)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;

        SETUP_PID;

        /* see if the front end will allow us to perform this
           function asynchronously.  */
        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return smb_raw_unlink(private->tree, unl);
        }

        c_req = smb_raw_unlink_send(private->tree, unl);

        SIMPLE_ASYNC_TAIL;
}

/*
  a handler for async ioctl replies
 */
static void async_ioctl(struct smbcli_request *c_req)
{
        struct async_info *async = c_req->async.private;
        struct ntvfs_request *req = async->req;
        req->async_states->status = smb_raw_ioctl_recv(c_req, req, async->parms);
        talloc_free(async);
        req->async_states->send_fn(req);
}

/*
  ioctl interface
*/
static NTSTATUS proxy_ioctl(struct ntvfs_module_context *ntvfs,
                           struct ntvfs_request *req, union smb_ioctl *io)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;

        if (io->ntioctl.level == RAW_IOCTL_NTIOCTL
                && io->ntioctl.in.function == FSCTL_UFOPROXY_RPCLITE) {
                return proxy_rpclite(ntvfs, req, io);
        }

        SETUP_PID_AND_FILE;

        /* see if the front end will allow us to perform this
           function asynchronously.  */
        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return smb_raw_ioctl(private->tree, req, io);
        }

        c_req = smb_raw_ioctl_send(private->tree, io);

        ASYNC_RECV_TAIL(io, async_ioctl);
}

/*
  check if a directory exists
*/
static NTSTATUS proxy_chkpath(struct ntvfs_module_context *ntvfs,
                             struct ntvfs_request *req, union smb_chkpath *cp)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;

        SETUP_PID;

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return smb_raw_chkpath(private->tree, cp);
        }

        c_req = smb_raw_chkpath_send(private->tree, cp);

        SIMPLE_ASYNC_TAIL;
}

static bool find_search_cache_item(const char* path,
                                   struct search_cache **search_cache,
                                   struct search_cache_item **item) {
        struct search_cache *s=*search_cache;
        struct search_cache_item *i=*item;
        const char* file;
        int dir_len;

        /* see if we can satisfy from a directory cache */
        DEBUG(5,("%s: Looking for pathinfo: '%s'\n",__LOCATION__,path));
        if ((file=strrchr(path,'\\'))) {
                dir_len = file - path;
                /* point past the \ */
                file++;
        } else {
                file = path;
                dir_len = 0;
        }
        /* convert empty path to . so we can find it in the cache */
        if (! *file) {
                file=".";
        }
        DEBUG(5,("%s: Path='%s' File='%s'\n",__LOCATION__,path, file));

        /* Note we don't care if the cache is partial, as long as it has a hit */
        while(s) {
                /* One day we may support all directory levels */
                DLIST_FIND(s, s, (s->key.data_level == RAW_SEARCH_DATA_BOTH_DIRECTORY_INFO &&
                                  strlen(s->dir)==dir_len &&
                                  fstrncmp(s->dir, path, dir_len)==0));
                if (! s) {
                        break;
                }
                DEBUG(5,("%s: found cache %p\n",__LOCATION__,s));
                /* search s for io->generic.in.file.path */
                DLIST_FIND(s->items, i, (i->data_level == RAW_SEARCH_DATA_BOTH_DIRECTORY_INFO &&
                                         ((i->file->both_directory_info.name.s &&
                                           fstrcmp(i->file->both_directory_info.name.s, file) ==0) ||
                                          (i->file->both_directory_info.short_name.s &&
                                           fstrcmp(i->file->both_directory_info.short_name.s, file)==0)
                                          )));
                DEBUG(5,("%s: found cache %p item %p\n",__LOCATION__,s, i));
                if (i) {
                        *item=i;
                        *search_cache=s;
                        return true;
                }
                s=s->next;
                DEBUG(5,("%s: continue search at %p\n",__LOCATION__,s));
        }
        *item=i;
        *search_cache=s;
        return false;
}

static void proxy_set_cache_info(struct file_metadata *metadata, struct proxy_GetInfo *r) {
        /* only set this if it was responded... I think they all are responded... */
        metadata->info_data.status_RAW_FILEINFO_BASIC_INFORMATION=r->out.info_data[0].status_RAW_FILEINFO_BASIC_INFORMATION;
        if (NT_STATUS_IS_OK(r->out.info_data[0].status_RAW_FILEINFO_BASIC_INFORMATION) /*||
            /*NT_STATUS_IS_OK(r->out.info_data[0].status_RAW_FILEINFO_ALL_INFO)*/) {
                metadata->info_data.create_time=r->out.info_data[0].create_time;
                metadata->info_data.access_time =r->out.info_data[0].access_time;
                metadata->info_data.write_time=r->out.info_data[0].write_time;
                metadata->info_data.change_time=r->out.info_data[0].change_time;
                metadata->info_data.attrib=r->out.info_data[0].attrib;
                metadata->valid|=valid_RAW_FILEINFO_BASIC_INFORMATION;
        }
        metadata->info_data.status_RAW_FILEINFO_ALL_INFO=r->out.info_data[0].status_RAW_FILEINFO_ALL_INFO;
        if (NT_STATUS_IS_OK(r->out.info_data[0].status_RAW_FILEINFO_ALL_INFO)) {
                metadata->info_data.ea_size=r->out.info_data[0].ea_size;
                metadata->info_data.alloc_size=r->out.info_data[0].alloc_size;
                metadata->info_data.size=r->out.info_data[0].size;
                metadata->info_data.nlink=r->out.info_data[0].nlink;
                /* Are we duping this right? Would talloc_reference be ok? */
                //f->metadata->info_data.fname=
                metadata->info_data.fname.s=talloc_memdup(metadata, r->out.info_data[0].fname.s, r->out.info_data[0].fname.count);
                metadata->info_data.fname.count=r->out.info_data[0].fname.count;
                metadata->info_data.delete_pending=r->out.info_data[0].delete_pending;
                metadata->info_data.directory=r->out.info_data[0].directory;
                metadata->valid|=valid_RAW_FILEINFO_ALL_INFO | valid_RAW_FILEINFO_STANDARD_INFO;;
        }
        metadata->info_data.status_RAW_FILEINFO_COMPRESSION_INFO=r->out.info_data[0].status_RAW_FILEINFO_COMPRESSION_INFO;
        if (NT_STATUS_IS_OK(r->out.info_data[0].status_RAW_FILEINFO_COMPRESSION_INFO)) {
                metadata->info_data.compressed_size=r->out.info_data[0].compressed_size;
                metadata->info_data.format=r->out.info_data[0].format;
                metadata->info_data.unit_shift=r->out.info_data[0].unit_shift;
                metadata->info_data.chunk_shift=r->out.info_data[0].chunk_shift;
                metadata->info_data.cluster_shift=r->out.info_data[0].cluster_shift;
                metadata->valid|=valid_RAW_FILEINFO_COMPRESSION_INFO;
        }
        metadata->info_data.status_RAW_FILEINFO_INTERNAL_INFORMATION=r->out.info_data[0].status_RAW_FILEINFO_INTERNAL_INFORMATION;
        if (NT_STATUS_IS_OK(r->out.info_data[0].status_RAW_FILEINFO_INTERNAL_INFORMATION)) {
                metadata->info_data.file_id=r->out.info_data[0].file_id;
                metadata->valid|=valid_RAW_FILEINFO_INTERNAL_INFORMATION;
        }
        metadata->info_data.status_RAW_FILEINFO_ACCESS_INFORMATION=r->out.info_data[0].status_RAW_FILEINFO_ACCESS_INFORMATION;
        if (NT_STATUS_IS_OK(r->out.info_data[0].status_RAW_FILEINFO_ACCESS_INFORMATION)) {
                metadata->info_data.access_flags=r->out.info_data[0].access_flags;
                metadata->valid|=valid_RAW_FILEINFO_ACCESS_INFORMATION;
        }
        metadata->info_data.status_RAW_FILEINFO_POSITION_INFORMATION=r->out.info_data[0].status_RAW_FILEINFO_POSITION_INFORMATION;
        if (NT_STATUS_IS_OK(r->out.info_data[0].status_RAW_FILEINFO_POSITION_INFORMATION)) {
                metadata->info_data.position=r->out.info_data[0].position;
                metadata->valid|=valid_RAW_FILEINFO_POSITION_INFORMATION;
        }
        metadata->info_data.status_RAW_FILEINFO_MODE_INFORMATION=r->out.info_data[0].status_RAW_FILEINFO_MODE_INFORMATION;
        if (NT_STATUS_IS_OK(r->out.info_data[0].status_RAW_FILEINFO_MODE_INFORMATION)) {
                metadata->info_data.mode=r->out.info_data[0].mode;
                metadata->valid|=valid_RAW_FILEINFO_MODE_INFORMATION;
        }
        metadata->info_data.status_RAW_FILEINFO_ALIGNMENT_INFORMATION=r->out.info_data[0].status_RAW_FILEINFO_ALIGNMENT_INFORMATION;
        if (NT_STATUS_IS_OK(r->out.info_data[0].status_RAW_FILEINFO_ALIGNMENT_INFORMATION)) {
                metadata->info_data.alignment_requirement=r->out.info_data[0].alignment_requirement;
                metadata->valid|=valid_RAW_FILEINFO_ALIGNMENT_INFORMATION;
        }
        metadata->info_data.status_RAW_FILEINFO_ATTRIBUTE_TAG_INFORMATION=r->out.info_data[0].status_RAW_FILEINFO_ATTRIBUTE_TAG_INFORMATION;
        if (NT_STATUS_IS_OK(r->out.info_data[0].status_RAW_FILEINFO_ATTRIBUTE_TAG_INFORMATION)) {
                metadata->info_data.reparse_tag=r->out.info_data[0].reparse_tag;
                metadata->info_data.reparse_attrib=r->out.info_data[0].reparse_attrib;
                metadata->valid|=valid_RAW_FILEINFO_ATTRIBUTE_TAG_INFORMATION;
        }
        metadata->info_data.status_RAW_FILEINFO_STREAM_INFO=r->out.info_data[0].status_RAW_FILEINFO_STREAM_INFO;
        if (NT_STATUS_IS_OK(r->out.info_data[0].status_RAW_FILEINFO_STREAM_INFO)) {
                metadata->info_data.num_streams=r->out.info_data[0].num_streams;
                talloc_free(metadata->info_data.streams);
                metadata->info_data.streams=talloc_steal(metadata, r->out.info_data[0].streams);
                metadata->valid|=valid_RAW_FILEINFO_STREAM_INFO;
        }
}
/* satisfy a file-info request from cache */
NTSTATUS proxy_cache_info(union smb_fileinfo *io, struct file_metadata *metadata, bool *valid)
{
#define SET_VALID(FLAG) do { \
if (valid) *valid=!!(metadata->valid & valid_ ## FLAG); \
DEBUG(5,("%s check %s=%d (%x)\n",__FUNCTION__, #FLAG, !!(metadata->valid & valid_ ## FLAG), metadata->valid)); \
} while(0)
        /* and now serve the request from the cache */
        switch(io->generic.level) {
        case RAW_FILEINFO_BASIC_INFORMATION:
                SET_VALID(RAW_FILEINFO_BASIC_INFORMATION);
                io->basic_info.out.create_time=metadata->info_data.create_time;
                io->basic_info.out.access_time=metadata->info_data.access_time;
                io->basic_info.out.write_time=metadata->info_data.write_time;
                io->basic_info.out.change_time=metadata->info_data.change_time;
                io->basic_info.out.attrib=metadata->info_data.attrib;
                return metadata->info_data.status_RAW_FILEINFO_BASIC_INFORMATION;
        case RAW_FILEINFO_ALL_INFO:
                SET_VALID(RAW_FILEINFO_ALL_INFO);
                io->all_info.out.create_time=metadata->info_data.create_time;
                io->all_info.out.access_time=metadata->info_data.access_time;
                io->all_info.out.write_time=metadata->info_data.write_time;
                io->all_info.out.change_time=metadata->info_data.change_time;
                io->all_info.out.attrib=metadata->info_data.attrib;
                io->all_info.out.alloc_size=metadata->info_data.alloc_size;
                io->all_info.out.size=metadata->info_data.size;
                io->all_info.out.directory=metadata->info_data.directory;
                io->all_info.out.nlink=metadata->info_data.nlink;
                io->all_info.out.delete_pending=metadata->info_data.delete_pending;
                io->all_info.out.fname.s=metadata->info_data.fname.s;
                io->all_info.out.fname.private_length=metadata->info_data.fname.count;
                return metadata->info_data.status_RAW_FILEINFO_ALL_INFO;
        case RAW_FILEINFO_STANDARD_INFO:
        case RAW_FILEINFO_STANDARD_INFORMATION:
                SET_VALID(RAW_FILEINFO_ALL_INFO);
                io->standard_info.out.alloc_size=metadata->info_data.alloc_size;
                io->standard_info.out.size=metadata->info_data.size;
                io->standard_info.out.directory=metadata->info_data.directory;
                io->standard_info.out.nlink=metadata->info_data.nlink; /* may be wrong */
                io->standard_info.out.delete_pending=metadata->info_data.delete_pending;
                return metadata->info_data.status_RAW_FILEINFO_ALL_INFO;
        case RAW_FILEINFO_EA_INFO:
        case RAW_FILEINFO_EA_INFORMATION:
                SET_VALID(RAW_FILEINFO_ALL_INFO);
                io->ea_info.out.ea_size=metadata->info_data.ea_size;
                return metadata->info_data.status_RAW_FILEINFO_ALL_INFO;
        case RAW_FILEINFO_COMPRESSION_INFO:
                SET_VALID(RAW_FILEINFO_COMPRESSION_INFO);
                io->compression_info.out.compressed_size=metadata->info_data.compressed_size;
                io->compression_info.out.format=metadata->info_data.format;
                io->compression_info.out.unit_shift=metadata->info_data.unit_shift;
                io->compression_info.out.chunk_shift=metadata->info_data.chunk_shift;
                io->compression_info.out.cluster_shift=metadata->info_data.cluster_shift;
                return metadata->info_data.status_RAW_FILEINFO_COMPRESSION_INFO;
        case RAW_FILEINFO_INTERNAL_INFORMATION:
                SET_VALID(RAW_FILEINFO_INTERNAL_INFORMATION);
                io->internal_information.out.file_id=metadata->info_data.file_id;
                return metadata->info_data.status_RAW_FILEINFO_INTERNAL_INFORMATION;
        case RAW_FILEINFO_ACCESS_INFORMATION:
                SET_VALID(RAW_FILEINFO_ACCESS_INFORMATION);
                io->access_information.out.access_flags=metadata->info_data.access_flags;
                return metadata->info_data.status_RAW_FILEINFO_ACCESS_INFORMATION;
        case RAW_FILEINFO_POSITION_INFORMATION:
                SET_VALID(RAW_FILEINFO_POSITION_INFORMATION);
                io->position_information.out.position=metadata->info_data.position;
                return metadata->info_data.status_RAW_FILEINFO_POSITION_INFORMATION;
        case RAW_FILEINFO_MODE_INFORMATION:
                SET_VALID(RAW_FILEINFO_MODE_INFORMATION);
                io->mode_information.out.mode=metadata->info_data.mode;
                return metadata->info_data.status_RAW_FILEINFO_MODE_INFORMATION;
        case RAW_FILEINFO_ALIGNMENT_INFORMATION:
                SET_VALID(RAW_FILEINFO_ALIGNMENT_INFORMATION);
                io->alignment_information.out.alignment_requirement=metadata->info_data.alignment_requirement;
                return metadata->info_data.status_RAW_FILEINFO_ALIGNMENT_INFORMATION;
        case RAW_FILEINFO_ATTRIBUTE_TAG_INFORMATION:
                SET_VALID(RAW_FILEINFO_ATTRIBUTE_TAG_INFORMATION);
                io->attribute_tag_information.out.reparse_tag=metadata->info_data.reparse_tag;
                io->attribute_tag_information.out.attrib=metadata->info_data.reparse_attrib;
                return metadata->info_data.status_RAW_FILEINFO_ATTRIBUTE_TAG_INFORMATION;
        case RAW_FILEINFO_STREAM_INFO:
        case RAW_FILEINFO_STREAM_INFORMATION:
                SET_VALID(RAW_FILEINFO_STREAM_INFO);
                io->stream_info.out.num_streams=metadata->info_data.num_streams;
                if (metadata->info_data.num_streams > 0) {
                        io->stream_info.out.streams = talloc_zero_array(io, struct stream_struct, metadata->info_data.num_streams);
                        int c;
                        if (! io->stream_info.out.streams) {
                                if (*valid) *valid=false;
                                io->stream_info.out.num_streams=0;
                                return NT_STATUS_NO_MEMORY;
                        }
                        for (c=0; c<io->stream_info.out.num_streams; c++) {
                                io->stream_info.out.streams[c].size = metadata->info_data.streams[c].size;
                                io->stream_info.out.streams[c].alloc_size = metadata->info_data.streams[c].alloc_size;
                                io->stream_info.out.streams[c].stream_name.s = talloc_reference(io, metadata->info_data.streams[c].stream_name.s);
                                io->stream_info.out.streams[c].stream_name.private_length = metadata->info_data.streams[c].stream_name.count;
                        }
                } else {
                        io->stream_info.out.streams=NULL;
                }
                return metadata->info_data.status_RAW_FILEINFO_STREAM_INFO;
        default:
                DEBUG(5,("%s: Unknown request\n",__FUNCTION__));
                if (valid) *valid=false;
                return NT_STATUS_INTERNAL_ERROR;
        }
}

/*
  a handler for async qpathinfo replies
 */
static void async_qpathinfo(struct smbcli_request *c_req)
{
        struct async_info *async = c_req->async.private;
        struct ntvfs_request *req = async->req;
        req->async_states->status = smb_raw_pathinfo_recv(c_req, req, async->parms);
        talloc_free(async);
        req->async_states->send_fn(req);
}

static NTSTATUS async_proxy_qpathinfo(struct async_info *async, void* io1, void* io2, NTSTATUS status)
{
        struct proxy_private *private = async->proxy;
        struct smbcli_request *c_req = async->c_req;
        struct ntvfs_request *req = async->req;
        struct proxy_file *f = talloc_get_type_abort(async->f, struct proxy_file);
        union smb_fileinfo *io = talloc_get_type_abort(io1, union smb_fileinfo);
        struct proxy_GetInfo *r=talloc_get_type_abort(io2, struct proxy_GetInfo);

        if (c_req) status = smb_raw_fileinfo_recv(c_req, req, async->parms);
        req->async_states->status=status;

        /* It's good to check for over-all status but we need to check status of each sub-message */
        NT_STATUS_NOT_OK_RETURN(status);

        /* populate the cache, and then fill the request from the cache */
        /* Assuming that r->count.in == 1 */
        SMB_ASSERT(r->out.count==1);
        DEBUG(5,("%s: Combined status of meta request: %s\n",__LOCATION__, get_friendly_nt_error_msg (r->out.info_data[0].status)));
        NT_STATUS_NOT_OK_RETURN(r->out.info_data[0].status);

        DEBUG(5,("%s: will set cache %p item=%p metadata=%p %p\n",__LOCATION__, f, f?f->metadata:NULL, r));
        proxy_set_cache_info(f->metadata, r);

        req->async_states->status=proxy_cache_info(io, f->metadata, NULL);
        DEBUG(5,("%s: set final response of original request to: %s\n",__LOCATION__, get_friendly_nt_error_msg (req->async_states->status)));

        return req->async_states->status;
}

static void async_qpathinfo_notify(void* data, struct fdirmon* dirmon) {
        struct proxy_file* file=data;

        DEBUG(5,("%s: qpathinfo cache %s destroyed\n",__LOCATION__,file->filename));
        DLIST_REMOVE(file->proxy->closed_files, file);
        talloc_free(file);
}

/*
  return info on a pathname
*/
static NTSTATUS proxy_qpathinfo(struct ntvfs_module_context *ntvfs,
                               struct ntvfs_request *req, union smb_fileinfo *io)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;
        struct proxy_file *f=NULL;
        const char* path;

        SETUP_PID;

        /* Look for closed files */
        if (private->enabled_qpathinfo) {
                int len=strlen(io->generic.in.file.path)+1;
                DEBUG(5,("%s: Looking for cached metadata for: %s\n",__LOCATION__,io->generic.in.file.path));
                DLIST_FIND(private->closed_files, f,
                           (len==f->filename_size && fstrncmp(io->generic.in.file.path, f->filename, f->filename_size)==0));
                if (f) {
                        /* stop cache going away while we are using it */
                        talloc_reference(req, f);
                }
        }
        /* upgrade the request */
        switch(io->generic.level) {
        case RAW_FILEINFO_STANDARD_INFO:
        case RAW_FILEINFO_STANDARD_INFORMATION:
        case RAW_FILEINFO_BASIC_INFORMATION: /* we get this on file open */
        case RAW_FILEINFO_ALL_INFO:
        case RAW_FILEINFO_COMPRESSION_INFO:
        case RAW_FILEINFO_INTERNAL_INFORMATION:
        case RAW_FILEINFO_ACCESS_INFORMATION:
        case RAW_FILEINFO_POSITION_INFORMATION:
        case RAW_FILEINFO_MODE_INFORMATION:
        case RAW_FILEINFO_ALIGNMENT_INFORMATION:
        case RAW_FILEINFO_ATTRIBUTE_TAG_INFORMATION:
        case RAW_FILEINFO_STREAM_INFO:
        case RAW_FILEINFO_STREAM_INFORMATION:
        case RAW_FILEINFO_EA_INFO:
        case RAW_FILEINFO_EA_INFORMATION:
                DEBUG(5,("%s: item is %p\n",__FUNCTION__, f));
                if (f && f->metadata) {
                        NTSTATUS status;
                        bool valid;
                        DEBUG(5,("%s: Using cached metadata %x (item=%p)\n",__FUNCTION__, f->metadata->valid, f));
                        status=proxy_cache_info(io, f->metadata, &valid);
                        if (valid) return status;
                        DEBUG(5,("%s: But cached metadata not valid :-(\n",__FUNCTION__));
                }
                /* construct an item to hold the cache if we need to */
                if (! f && private->enabled_cache_info && PROXY_REMOTE_SERVER(private) && (f=talloc_zero(private, struct proxy_file))) {
                        struct fdirmon* dirmon;
                        dirmon=get_fdirmon(private, io->generic.in.file.path, true);
                        if (f && dirmon) {
                                f->proxy=private;
                                dirmon_add_callback(dirmon, async_qpathinfo_notify, f);

                                f->filename=talloc_strdup(f, io->generic.in.file.path);
                                f->filename_size=strlen(f->filename)+1;
                                f->metadata=talloc_zero(f, struct file_metadata);
                                /* should not really add unless we succeeded */
                                DLIST_ADD(private->closed_files, f);
                        } else {
                                talloc_free(f);
                                f=NULL;
                        }
                }
                if (f && f->metadata && private->enabled_cache_info && PROXY_REMOTE_SERVER(private)) {
                        struct proxy_GetInfo *r;
                        DEBUG(5,("%s: promoting request to proxy\n",__FUNCTION__));

                        r=talloc_zero(req, struct proxy_GetInfo);
                        NT_STATUS_HAVE_NO_MEMORY(r);

                        r->in.count=1;
                        r->in.info_tags=talloc_zero_array(req, struct info_tags, r->in.count);
                        r->in.info_tags[0].tag_type=TAG_TYPE_PATH_INFO;
                        /* 1+ to get the null */
                        r->in.info_tags[0].info_tag.path.count=1+strlen(io->generic.in.file.path);
                        r->in.info_tags[0].info_tag.path.s=io->generic.in.file.path;
                        c_req = smbcli_ndr_request_ntioctl_send(private->tree, ntvfs, &ndr_table_rpcproxy, NDR_PROXY_GETINFO, r);
                        /* the callback handler will populate the cache and respond from the cache */
                        ADD_ASYNC_RECV_TAIL(c_req, io, r, f, async_proxy_qpathinfo, NT_STATUS_INTERNAL_ERROR);

                        if (! (req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                                DEBUG(5,("%s Sync waiting promotion\n",__FUNCTION__));
                                return sync_chain_handler(c_req);
                        } else {
                                void* f=NULL;
                                DEBUG(5,("%s Async waiting promotion\n",__FUNCTION__));
                                ASYNC_RECV_TAIL_HANDLER_ORPHAN(io, async_chain_handler);
                                req->async_states->state |= NTVFS_ASYNC_STATE_ASYNC;
                                return NT_STATUS_OK;
                        }
                }
        }

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return smb_raw_pathinfo(private->tree, req, io);
        }

        c_req = smb_raw_pathinfo_send(private->tree, io);

        ASYNC_RECV_TAIL(io, async_qpathinfo);
}

/*
  a handler for async qfileinfo replies
 */
static void async_qfileinfo(struct smbcli_request *c_req)
{
        struct async_info *async = c_req->async.private;
        struct ntvfs_request *req = async->req;
        req->async_states->status = smb_raw_fileinfo_recv(c_req, req, async->parms);
        talloc_free(async);
        req->async_states->send_fn(req);
}

static NTSTATUS async_proxy_qfileinfo(struct async_info *async, void* io1, void* io2, NTSTATUS status)
{
        struct proxy_private *private = async->proxy;
        struct smbcli_request *c_req = async->c_req;
        struct ntvfs_request *req = async->req;
        struct proxy_file *f = async->f;
        union smb_fileinfo *io = talloc_get_type_abort(io1, union smb_fileinfo);
        struct proxy_GetInfo *r=talloc_get_type_abort(io2, struct proxy_GetInfo);

        if (c_req) status = smb_raw_fileinfo_recv(c_req, req, async->parms);
        req->async_states->status=status;

        NT_STATUS_NOT_OK_RETURN(status);

        /* populate the cache, and then fill the request from the cache */
        /* Assuming that r->count.in == 1 */
        SMB_ASSERT(r->out.count==1);
        NT_STATUS_NOT_OK_RETURN(r->out.info_data[0].status);

        proxy_set_cache_info(f->metadata, r);

        req->async_states->status=proxy_cache_info(io, f->metadata, NULL);

        return req->async_states->status;
}

/*
  query info on a open file
*/
static NTSTATUS proxy_qfileinfo(struct ntvfs_module_context *ntvfs,
                               struct ntvfs_request *req, union smb_fileinfo *io)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;
        struct proxy_file *f;
        bool valid=false;
        NTSTATUS status;

        SETUP_PID;

        SETUP_FILE_HERE(f);

        /* upgrade the request */
        switch(io->generic.level) {
        case RAW_FILEINFO_STANDARD_INFO:
        case RAW_FILEINFO_STANDARD_INFORMATION:
        case RAW_FILEINFO_BASIC_INFORMATION: /* we get this on file open */
        case RAW_FILEINFO_ALL_INFO:
        case RAW_FILEINFO_COMPRESSION_INFO:
        case RAW_FILEINFO_INTERNAL_INFORMATION:
        case RAW_FILEINFO_ACCESS_INFORMATION:
        case RAW_FILEINFO_POSITION_INFORMATION:
        case RAW_FILEINFO_MODE_INFORMATION:
        case RAW_FILEINFO_ALIGNMENT_INFORMATION:
        case RAW_FILEINFO_ATTRIBUTE_TAG_INFORMATION:
        case RAW_FILEINFO_STREAM_INFO:
        case RAW_FILEINFO_STREAM_INFORMATION:
        case RAW_FILEINFO_EA_INFO:
        case RAW_FILEINFO_EA_INFORMATION:
                DEBUG(5,("%s: oplock is %d\n",__FUNCTION__, f->oplock));
                if (f->oplock) {
                        DEBUG(5,("%s: %p Using cached metadata %x (fnum=%d)\n",__FUNCTION__, f, f->metadata->valid, f->fnum));
                        status=proxy_cache_info(io, f->metadata, &valid);
                        if (valid) return status;
                        DEBUG(5,("%s: But cached metadata not valid :-(\n",__FUNCTION__));
                }
                if (private->enabled_cache_info && PROXY_REMOTE_SERVER(private)) {
                        DEBUG(5,("%s: promoting request to proxy\n",__FUNCTION__));
                        struct proxy_GetInfo *r=talloc_zero(req, struct proxy_GetInfo);
                        NT_STATUS_HAVE_NO_MEMORY(r);
                        r->in.count=1;
                        r->in.info_tags=talloc_zero_array(req, struct info_tags, r->in.count);
                        r->in.info_tags[0].tag_type=TAG_TYPE_FILE_INFO;
                        r->in.info_tags[0].info_tag.fnum=io->generic.in.file.fnum;
                        c_req = smbcli_ndr_request_ntioctl_send(private->tree, ntvfs, &ndr_table_rpcproxy, NDR_PROXY_GETINFO, r);
                        /* the callback handler will populate the cache and respond from the cache */
                        ADD_ASYNC_RECV_TAIL(c_req, io, r, f, async_proxy_qfileinfo, NT_STATUS_INTERNAL_ERROR);

                        if (! (req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                                DEBUG(5,("%s Sync waiting promotion\n",__FUNCTION__));
                                return sync_chain_handler(c_req);
                        } else {
                                DEBUG(5,("%s Async waiting promotion\n",__FUNCTION__));
                                ASYNC_RECV_TAIL_HANDLER_ORPHAN(io, async_chain_handler);
                                req->async_states->state |= NTVFS_ASYNC_STATE_ASYNC;
                                return NT_STATUS_OK;
                        }
                }
        }

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return smb_raw_fileinfo(private->tree, req, io);
        }

        c_req = smb_raw_fileinfo_send(private->tree, io);

        ASYNC_RECV_TAIL(io, async_qfileinfo);
}

/*
  set info on a pathname
*/
static NTSTATUS proxy_setpathinfo(struct ntvfs_module_context *ntvfs,
                                 struct ntvfs_request *req, union smb_setfileinfo *st)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;

        SETUP_PID;

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return smb_raw_setpathinfo(private->tree, st);
        }

        c_req = smb_raw_setpathinfo_send(private->tree, st);

        SIMPLE_ASYNC_TAIL;
}


/*
  a handler for async open replies
 */
static void async_open(struct smbcli_request *c_req)
{
        struct async_info *async = c_req->async.private;
        struct proxy_private *proxy = async->proxy;
        struct ntvfs_request *req = async->req;
        struct proxy_file *f = async->f;
        union smb_open *io = async->parms;
        union smb_handle *file;

        talloc_free(async);
        req->async_states->status = smb_raw_open_recv(c_req, req, io);
        SMB_OPEN_OUT_FILE(io, file);
        f->fnum = file->fnum;
        file->ntvfs = NULL;
        if (!NT_STATUS_IS_OK(req->async_states->status)) goto failed;
        req->async_states->status = ntvfs_handle_set_backend_data(f->h, proxy->ntvfs, f);
        if (!NT_STATUS_IS_OK(req->async_states->status)) goto failed;
        file->ntvfs = f->h;
        DLIST_ADD(proxy->files, f);

        f->oplock=io->generic.out.oplock_level;

        LOAD_CACHE_FILE_DATA (f->metadata->info_data, io->generic.out);
        f->metadata->valid |= valid_RAW_FILEINFO_BASIC_INFORMATION;
    DEBUG(5,("**** METADATA VALID %p %x LEN=%lld\n",f,f->metadata->valid,(long long)f->metadata->info_data.size));

        if (proxy->cache_enabled) {
                struct search_cache_item *item=NULL;
                struct search_cache *s=proxy->search_caches;
                /* If we are still monitoring the file for changes we can
                   retain the previous cache state, [if it is more recent that the monitor]! */
                /* yeah yeah what if there is more than one.... :-( */
                if (! (io->generic.level == RAW_OPEN_NTCREATEX &&
                     io->generic.in.create_options & NTCREATEX_OPTIONS_OPEN_BY_FILE_ID) &&
                    find_search_cache_item(SMB_OPEN_IN_FILE(io), &s, &item) && item->cache) {
                        DEBUG(5,("%s: Using cached file cache\n",__LOCATION__));
                        f->cache=talloc_reference(f, item->cache);
                        cache_beopen(f->cache);
                        if (item->metadata) {
                                *(f->metadata)=*(item->metadata);
                                f->metadata->info_data.fname.s=talloc_strdup(f, item->metadata->info_data.fname.s);
                                f->metadata->info_data.fname.count=item->metadata->info_data.fname.count;

                                f->metadata->info_data.streams=talloc_zero_array(f, struct info_stream, f->metadata->info_data.num_streams);
                                if (f->metadata->info_data.streams) {
                                        int c;
                                        for(c=0; c < f->metadata->info_data.num_streams; c++) {
                                                f->metadata->info_data.streams[c].size = item->metadata->info_data.streams[c].size;
                                                f->metadata->info_data.streams[c].alloc_size = item->metadata->info_data.streams[c].alloc_size;
                                                f->metadata->info_data.streams[c].stream_name.s= talloc_strdup(f, item->metadata->info_data.streams[c].stream_name.s);
                                                f->metadata->info_data.streams[c].stream_name.count=item->metadata->info_data.streams[c].stream_name.count;
                                        }
                                }
                                f->metadata->count=1;
                        }
                } else {
                        f->cache=cache_open(proxy->cache, f, io, f->oplock, proxy->cache_readahead);
                        if (proxy->fake_valid) {
                                cache_handle_validated(f, cache_handle_len(f));
                        }
                        if (! PROXY_REMOTE_SERVER(proxy)) cache_handle_novalidate(f);
                        if (item) {
                                item->cache = talloc_reference(item, f->cache);
                                item->metadata=talloc_reference(item, f->metadata);
                                DEBUG(5,("%s: Caching file cache for later\n",__LOCATION__));
                        } else {
                                DEBUG(5,("%s: NOT Caching file cache for later\n",__LOCATION__));
                        }
                }
        }

failed:
        req->async_states->send_fn(req);
}

/*
  open a file
*/
static NTSTATUS proxy_open(struct ntvfs_module_context *ntvfs,
                          struct ntvfs_request *req, union smb_open *io)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;
        struct ntvfs_handle *h;
        struct proxy_file *f, *clone;
        NTSTATUS status;
        void *filename;
        int filename_size;
        uint16_t fnum;

        SETUP_PID;

        if (io->generic.level != RAW_OPEN_GENERIC &&
            private->map_generic) {
                return ntvfs_map_open(ntvfs, req, io);
        }

        status = ntvfs_handle_new(ntvfs, req, &h);
#warning should we free this handle if the open fails?
        NT_STATUS_NOT_OK_RETURN(status);

        f = talloc_zero(h, struct proxy_file);
        NT_STATUS_HAVE_NO_MEMORY(f);
        f->proxy=private;

        /* If the file is being opened read only and we already have a read-only
           handle for this file, then just clone and ref-count the handle */
        /* First calculate the filename key */
        if (io->generic.level == RAW_OPEN_NTCREATEX &&
            io->generic.in.create_options & NTCREATEX_OPTIONS_OPEN_BY_FILE_ID) {
                filename_size=sizeof(uint64_t);
                filename=io->generic.in.fname;
        } else {
                filename=SMB_OPEN_IN_FILE(io);
                filename_size=strlen(filename)+1;
        }
        f->filename=talloc_memdup(f, filename, filename_size);
        f->filename_size=filename_size;
        f->h = h;
        f->can_clone= (io->generic.in.access_mask & NTCREATEX_SHARE_ACCESS_MASK) == NTCREATEX_SHARE_ACCESS_READ &&
                      (io->generic.in.impersonation == NTCREATEX_IMPERSONATION_IMPERSONATION) &&
                      (io->generic.in.create_options & NTCREATEX_OPTIONS_DIRECTORY) == 0 &&
                      (io->generic.in.open_disposition != NTCREATEX_DISP_CREATE) &&
                      (io->generic.in.open_disposition != NTCREATEX_DISP_SUPERSEDE);
        /* see if we have a matching open file */
        clone=NULL;
        if (f->can_clone) for (clone=private->files; clone; clone=clone->next) {
                if (clone->can_clone && filename_size == clone->filename_size &&
                    memcmp(filename, clone->filename, filename_size)==0) {
                        break;
                }
        }

        /* if clone is not null, then we found a match */
        if (private->enabled_open_clone && clone) {
                union smb_handle *file;

                DEBUG(5,("%s: clone handle %d\n",__FUNCTION__,clone->fnum));
                SMB_OPEN_OUT_FILE(io, file);
                f->fnum = clone->fnum;
                file->ntvfs = NULL;
                status = ntvfs_handle_set_backend_data(f->h, private->ntvfs, f);
                NT_STATUS_NOT_OK_RETURN(status);
                file->ntvfs = f->h;
                DLIST_ADD(private->files, f);
                /* but be sure to share the same metadata cache */
                f->metadata=talloc_reference(f, clone->metadata);
                f->metadata->count++;
                f->oplock=clone->oplock;
                f->cache=talloc_reference(f, clone->cache);
                /* We don't need to reduce the oplocks for both files if we are read-only */
/*              if (clone->oplock==EXCLUSIVE_OPLOCK_RETURN ||
                    clone->oplock==BATCH_OPLOCK_RETURN) {
                        DEBUG(5,("%s: Breaking clone oplock from %d\n",__LOCATION__, clone->oplock));
                        clone->oplock==LEVEL_II_OPLOCK_RETURN;
                        status = ntvfs_send_oplock_break(private->ntvfs, clone->h, OPLOCK_BREAK_TO_LEVEL_II);
                        //if (!NT_STATUS_IS_OK(status)) result=false;
                } else if (clone->oplock==LEVEL_II_OPLOCK_RETURN) {
                        DEBUG(5,("%s: Breaking clone oplock from %d, cache no longer valid\n",__LOCATION__, clone->oplock));
                        cache_handle_stale(f);
                        clone->oplock=NO_OPLOCK_RETURN;
                        status = ntvfs_send_oplock_break(private->ntvfs, clone->h, OPLOCK_BREAK_TO_NONE);
                        //if (!NT_STATUS_IS_OK(status)) result=false;
                }
*/
                f->oplock=clone->oplock;
                /* and fake the rest of the response struct */
                io->generic.out.oplock_level=f->oplock;
                io->generic.out.create_action=NTCREATEX_ACTION_EXISTED;
                io->generic.out.create_time=f->metadata->info_data.create_time;
                io->generic.out.access_time=f->metadata->info_data.access_time;
                io->generic.out.write_time=f->metadata->info_data.write_time;
                io->generic.out.change_time=f->metadata->info_data.change_time;
                io->generic.out.attrib=f->metadata->info_data.attrib;
                io->generic.out.alloc_size=f->metadata->info_data.alloc_size;
                io->generic.out.size=f->metadata->info_data.size;
                io->generic.out.file_type=f->metadata->info_data.file_type;
                io->generic.out.ipc_state=f->metadata->info_data.ipc_state;
                io->generic.out.is_directory=f->metadata->info_data.is_directory;
                /* optional return values matching SMB2 tagged
                   values in the call */
                //io->generic.out.maximal_access;
                return NT_STATUS_OK;
        }
        f->metadata=talloc_zero(f, struct file_metadata);
        NT_STATUS_HAVE_NO_MEMORY(f->metadata);
        f->metadata->count=1;

        /* if oplocks aren't requested, optionally override and request them */
        if (! (io->generic.in.flags & (OPENX_FLAGS_REQUEST_OPLOCK | OPENX_FLAGS_REQUEST_BATCH_OPLOCK))
            && private->fake_oplock) {
            io->generic.in.flags |= OPENX_FLAGS_REQUEST_OPLOCK | OPENX_FLAGS_REQUEST_BATCH_OPLOCK;
        }

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                union smb_handle *file;

                status = smb_raw_open(private->tree, req, io);
                NT_STATUS_NOT_OK_RETURN(status);

                SMB_OPEN_OUT_FILE(io, file);
                f->fnum = file->fnum;
                file->ntvfs = NULL;
                status = ntvfs_handle_set_backend_data(f->h, private->ntvfs, f);
                NT_STATUS_NOT_OK_RETURN(status);
                file->ntvfs = f->h;
                DLIST_ADD(private->files, f);

                f->oplock=io->generic.out.oplock_level;

                LOAD_CACHE_FILE_DATA (f->metadata->info_data, io->generic.out);
    DEBUG(5,("**** METADATA VALID %p %x LEN=%lld\n",f,f->metadata->valid,(long long)f->metadata->info_data.size));
                f->metadata->valid |= valid_RAW_FILEINFO_BASIC_INFORMATION;

                if (private->cache_enabled) {
                        f->cache=cache_open(private->cache, f, io, f->oplock, private->cache_readahead);
                        if (private->fake_valid) {
                                cache_handle_validated(f, cache_handle_len(f));
                        }
                        if (! PROXY_REMOTE_SERVER(private)) cache_handle_novalidate(f);
                }

                return NT_STATUS_OK;
        }

        c_req = smb_raw_open_send(private->tree, io);

        ASYNC_RECV_TAIL_F(io, async_open, f);
}

/*
  create a directory
*/
static NTSTATUS proxy_mkdir(struct ntvfs_module_context *ntvfs,
                           struct ntvfs_request *req, union smb_mkdir *md)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;

        SETUP_PID;

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return smb_raw_mkdir(private->tree, md);
        }

        c_req = smb_raw_mkdir_send(private->tree, md);

        SIMPLE_ASYNC_TAIL;
}

/*
  remove a directory
*/
static NTSTATUS proxy_rmdir(struct ntvfs_module_context *ntvfs,
                           struct ntvfs_request *req, struct smb_rmdir *rd)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;

        SETUP_PID;

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return smb_raw_rmdir(private->tree, rd);
        }
        c_req = smb_raw_rmdir_send(private->tree, rd);

        SIMPLE_ASYNC_TAIL;
}

/*
  rename a set of files
*/
static NTSTATUS proxy_rename(struct ntvfs_module_context *ntvfs,
                            struct ntvfs_request *req, union smb_rename *ren)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;

        SETUP_PID;

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return smb_raw_rename(private->tree, ren);
        }

        c_req = smb_raw_rename_send(private->tree, ren);

        SIMPLE_ASYNC_TAIL;
}

/*
  copy a set of files
*/
static NTSTATUS proxy_copy(struct ntvfs_module_context *ntvfs,
                          struct ntvfs_request *req, struct smb_copy *cp)
{
        return NT_STATUS_NOT_SUPPORTED;
}

/* we only define this seperately so we can easily spot read calls in
   pending based on ( c_req->private.fn ==  async_read_handler ) */
static void async_read_handler(struct smbcli_request *c_req)
{
        async_chain_handler(c_req);
}

NTSTATUS async_readahead_dec(struct async_info *async, void* io1, void* io2, NTSTATUS status)
{
        struct proxy_private *private = async->proxy;
        struct smbcli_request *c_req = async->c_req;
        struct proxy_file *f = async->f;
        union smb_read *io = async->parms;

        /* if request is not already received by a chained handler, read it */
        if (c_req) status=smb_raw_read_recv(c_req, async->parms);

        DEBUG(3,("%s : file count %d, tree count %d\n",__FUNCTION__,
                         f->readahead_pending, private->readahead_spare));

        f->readahead_pending--;
        private->readahead_spare++;

        DEBUG(3,("%s : file count %d, tree count %d\n",__FUNCTION__,
                         f->readahead_pending, private->readahead_spare));

        return status;
}

/*
   a handler for async read replies - speculative read-aheads.
   It merely saves in the cache. The async chain handler will call send_fn if
   there is one, or if sync_chain_handler is used the send_fn is called by
   the ntvfs back end.
 */
NTSTATUS async_read_cache_save(struct async_info *async, void* io1, void* io2, NTSTATUS status)
{
        struct smbcli_request *c_req = async->c_req;
        struct proxy_file *f = async->f;
        union smb_read *io = async->parms;

        /* if request is not already received by a chained handler, read it */
        if (c_req) status=smb_raw_read_recv(c_req, async->parms);

        DEBUG(3,("%s async_read status: %s\n",__FUNCTION__,
                         get_friendly_nt_error_msg(status)));

        NT_STATUS_NOT_OK_RETURN(status);

        /* if it was a validate read we don't to save anything unless it failed.
           Until we use Proxy_read structs we can't tell, so guess */
        if (io->generic.out.nread == io->generic.in.maxcnt &&
                io->generic.in.mincnt < io->generic.in.maxcnt) {
                /* looks like a validate read, just move the validate pointer, the
                   original read-request has already been satisfied from cache */
                DEBUG(3,("%s megavalidate suceeded, validate to %lld\n",__FUNCTION__,
                                 io->generic.in.offset + io->generic.out.nread));
                cache_handle_validated(f, io->generic.in.offset + io->generic.out.nread);
        } else {
                DEBUG(5,("Not a mega-validate, save %d in cache\n",io->generic.out.nread));
                cache_handle_save(f, io->generic.out.data,
                                io->generic.out.nread,
                                io->generic.in.offset);
        }

        DEBUG(3,("%s finished %s\n",__FUNCTION__, get_friendly_nt_error_msg(status)));
        return status;
}

/* handler for fragmented reads */
NTSTATUS async_read_fragment(struct async_info *async, void* io1, void* io2, NTSTATUS status)
{
        struct proxy_private *private = async->proxy;
        struct smbcli_request *c_req = async->c_req;
        struct ntvfs_request *req = async->req;
        struct proxy_file *f = async->f;
        struct async_read_fragment* fragment=talloc_get_type_abort(io2, struct async_read_fragment);
        /* this is the io against which the fragment is to be applied */
        union smb_read *io = talloc_get_type_abort(io1, union smb_read);
        /* this is the io for the read that issued the callback */
        union smb_read *io_frag = fragment->io_frag; /* async->parms; */
        struct async_read_fragments* fragments=fragment->fragments;

        /* if request is not already received by a chained handler, read it */
#warning the queuer of the request should first push a suitable decoder, they should not scatter handlers generically
        if (c_req) status=smb_raw_read_recv(c_req, io_frag);

        DEBUG(3,("%s async_read status: %s\n",__FUNCTION__,
                         get_friendly_nt_error_msg(status)));

        fragment->status = status;

        /* remove fragment from fragments */
        DLIST_REMOVE(fragments->fragments, fragment);

#warning maybe read requests beyond the short read won't return NT_STATUS_OK with nread=0
/* in which case if we will want to collate all responses and return a valid read
   for the leading NT_STATUS_OK fragments */

        /* did this one fail, inducing a general fragments failure? */
        if (!NT_STATUS_IS_OK(fragment->status)) {
                /* preserve the status of the fragment with the smallest offset
                   when we can work out how */
                if (NT_STATUS_IS_OK(fragments->status)) {
                        fragments->status=fragment->status;
                }

                cache_handle_novalidate(f);
                DEBUG(5,("** Devalidated proxy due to read failure\n"));
        } else {
                /* No fragments have yet failed, keep collecting responses */
                ssize_t extent = io_frag->generic.in.offset + io_frag->generic.out.nread;
                /* Find memcpy window, copy data from the io_frag to the io */
                off_t start_offset=MAX(io_frag->generic.in.offset, io->generic.in.offset);
                /* used to use mincnt */
                off_t io_extent=io->generic.in.offset + io->generic.in.maxcnt;
                off_t end_offset=MIN(io_extent, extent);
                /* ASSERT(start_offset <= end_offset) */
                /* ASSERT(start_offset <= io_extent) */
                if (start_offset >= io_extent) {
                        DEBUG(3,("useless read-ahead tagged on to: %s",__LOCATION__));
                } else {
                        uint8_t* dst=io->generic.out.data+(start_offset - io->generic.in.offset);
                        uint8_t* src=io_frag->generic.out.data+(start_offset - io_frag->generic.in.offset);
                        /* src == dst in cases where we did not latch onto someone elses
                           read, but are handling our own */
                        if (src != dst)
                                memcpy(dst, src, end_offset - start_offset);
                }

                /* There should be a better way to detect, but it needs the proxy rpc struct
                   not ths smb_read struct */
                if (io_frag->generic.out.nread < io_frag->generic.in.maxcnt) {
                        DEBUG(5,("\n** Devalidated proxy due to small read: %lld min=%lld, max=%lld\n",
                                         (long long) io_frag->generic.out.nread,
                                         (long long) io_frag->generic.in.mincnt,
                                         (long long) io_frag->generic.in.maxcnt));
                        cache_handle_novalidate(f);
                }

                /* We broke up the original read. If not enough of this sub-read has
                   been read, and then some of then next block, it could leave holes!
                   We will only acknowledge up to the first partial read, and treat
                   it as a small read. If server can return NT_STATUS_OK for a partial
                   read so can we, so we preserve the response.
                   "enough" is all of it (maxcnt), except on the last block, when it has to
                   be enough to fill io->generic.in.mincnt. We know it is the last block
                   if nread is small but we could fill io->generic.in.mincnt */
                if (io_frag->generic.out.nread < io_frag->generic.in.mincnt &&
                        end_offset < io->generic.in.offset + io->generic.in.mincnt) {
                        DEBUG(4,("Fragmented read only partially successful\n"));

                        /* Shrink the master nread (or grow to this size if we are first partial */
                        if (! fragments->partial ||
                                        (io->generic.in.offset + io->generic.out.nread) > extent)  {
                                io->generic.out.nread = extent - io->generic.in.offset;
                        }

                        /* stop any further successes from extending the partial read */
                        fragments->partial=true;
                } else {
                        /* only grow the master nwritten if we haven't logged a partial write */
                        if (! fragments->partial &&
                                        (io->generic.in.offset + io->generic.out.nread) < extent ) {
                                io->generic.out.nread = MIN(io->generic.in.maxcnt, extent - io->generic.in.offset);
                        }
                }
        }

        /* Was it the last fragment, or do we know enought to send a response? */
        if (! fragments->fragments) {
                DEBUG(5,("Async read re-fragmented with %d of %d %s\n",
                                 io->generic.out.nread, io->generic.in.mincnt,
                                 get_friendly_nt_error_msg(fragments->status)));
                if (fragments->async) {
                        req->async_states->status=fragments->status;
                        DEBUG(5,("Fragments async response sending\n"));
#warning its not good freeing early if other pending requests have io allocated against this request which will now be freed
/* esp. as they may be attached to by other reads. Maybe attachees should be taking reference, but how will they
   know the top level they need to take reference too.. */
#warning should really queue a sender here, not call it */
        req->async_states->send_fn(req);
                        DEBUG(5,("Async response sent\n"));
                } else {
                        DEBUG(5,("Fragments SYNC return\n"));
                }
        }

        /* because a c_req may be shared by many req, chained handlers must return
           a status pertaining to the general validity of this specific c_req, not
           to their own private processing of the c_req for the benefit of their req
           which is returned in fragments->status
        */
        return status;
}

/* Issue read-ahead X bytes where X is the window size calculation based on
      server_latency * server_session_bandwidth
   where latency is the idle (link) latency and bandwidth is less than or equal_to
   to actual bandwidth available to the server.
   Read-ahead should honour locked areas in whatever way is neccessary (who knows?)
   read_ahead is defined here and not in the cache engine because it requires too
   much knowledge of private structures
*/
/* The concept is buggy unless we can tell the next proxy that these are
   read-aheads, otherwise chained proxy setups will each read-ahead of the
   read-ahead which can put a larger load on the final server.
   Also we probably need to distinguish between
    * cache-less read-ahead
    * cache-revalidating read-ahead
*/
NTSTATUS read_ahead(struct proxy_file *f, struct ntvfs_module_context *ntvfs,
                                        union smb_read *io, ssize_t as_read)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_tree *tree = private->tree;
        struct cache_file_entry *cache;
        off_t next_position; /* this read offset+length+window */
        off_t end_position; /* position we read-ahead to */
        off_t cache_populated;
        off_t read_position, new_extent;

        if (! PROXY_REMOTE_SERVER(private)) return NT_STATUS_UNSUCCESSFUL;
DEBUG(5,("A\n"));
        if (private->cache_readahead==0 || ! private->cache_enabled || ! f->cache) return NT_STATUS_UNSUCCESSFUL;
DEBUG(5,("B\n"));
        cache=talloc_get_type_abort(f->cache, struct cache_file_entry);
DEBUG(5,("C\n"));
        /* don't read-ahead if we are in bulk validate mode */
        if (cache->status & CACHE_VALIDATE) return NT_STATUS_UNSUCCESSFUL;
DEBUG(5,("D\n"));
        /* if we can't trust what we read-ahead anyway then don't bother although
         * if delta-reads are enabled we can do so in order to get something to
         * delta against */
        DEBUG(CACHE_DEBUG_LEVEL,("DOING Asking read-aheads: len %lld ra-extend %lld as-read %lld RA %d (%d)\n",
                                                         (long long int)(cache_len(cache)),
                                                         (long long int)(cache->readahead_extent),
                                                         (long long int)(as_read),
                                                         cache->readahead_window,private->cache_readahead));
        if (private->cache_readahead ==0 || ! (cache->status & CACHE_READ_AHEAD) ) {
                DEBUG(CACHE_DEBUG_LEVEL,("FAILED Asking read-aheads: Can't read-ahead as no read-ahead on this file: %x\n",
                                                                 cache->status));
                return NT_STATUS_UNSUCCESSFUL;
        }

        /* as_read is the mincnt bytes of a request being made or the
           out.nread of completed sync requests
           Here we presume that as_read bytes WILL be read. If there is a cache-ahead like ours,
           then this may often NOT be the case if readahead_window < requestsize; so we will
           get a small read, leaving a hole in the cache, and as we don't yet handle sparse caches,
           all future read-ahead will be wasted, so we need to adjust the read-ahead handler to handle
           this and have failed sparse writes adjust the cache->readahead_extent back to actual size */

        /* predict the file pointers next position */
        next_position=io->generic.in.offset + as_read;

        /* if we know how big the file is, don't read beyond */
        if (f->oplock && next_position > f->metadata->info_data.size) {
                next_position = f->metadata->info_data.size;
        }
        DEBUG(5,("Next position: %lld (%lld + %lld)\n",
                         (long long int)next_position,
                         (long long int)io->generic.in.offset,
                         (long long int)as_read));
        /* calculate the limit of the validated or requested cache */
        cache_populated=MAX(cache->validated_extent, cache->readahead_extent);

        /* will the new read take us beyond the current extent without gaps? */
        if (cache_populated < io->generic.in.offset) {
                /* this read-ahead is a read-behind-pointer */
                new_extent=cache_populated;
        } else {
                new_extent=MAX(next_position, cache_populated);
        }

        /* as far as we can tell new_extent is the smallest offset that doesn't
           have a pending read request on. Of course if we got a short read then
           we will have a cache-gap which we can't handle and need to read from
           a shrunk readahead_extent, which we don't currently handle */
        read_position=new_extent;

        /* of course if we know how big the remote file is we should limit at that */
        /* we should also mark-out which read-ahead requests are pending so that we
         * don't repeat them while they are in-transit.  */
        /* we can't really use next_position until we can have caches with holes
           UNLESS next_position < new_extent, because a next_position well before
           new_extent is no reason to extend it further, we only want to extended
           with read-aheads if we have cause to suppose the read-ahead data will
           be wanted, i.e. the next_position is near new_extent.
             So we can't justify reading beyond window+next_position, but if
           next_position is leaving gaps, we use new_extent instead */
        end_position=MIN(new_extent, next_position) + cache->readahead_window;
        if (f->oplock) {
                end_position=MIN(end_position, f->metadata->info_data.size);
        }
        DEBUG(5,("** Read-ahead loop %lld < %lld window=%d, end=%lld, quota: %d\n",
                         (long long int)read_position,
                         (long long int)(next_position + cache->readahead_window),
                         cache->readahead_window,
                         (long long int)end_position,
                         private->readahead_spare));
        /* do we even need to read? */
        if (! (read_position < end_position)) return NT_STATUS_OK;

        /* readahead_spare is for the whole session (mid/tid?) and may need sharing
           out over files and other tree-connects or something */
        while (read_position < end_position &&
                   private->readahead_spare > 0) {
                struct smbcli_request *c_req = NULL;
                ssize_t read_remaining = end_position - read_position;
                ssize_t read_block = MIN(private->tree->session->transport->negotiate.max_xmit - (MIN_SMB_SIZE+32),
                                         MIN(read_remaining, private->cache_readaheadblock));
                void *req = NULL; /* for the ASYNC_REC_TAIL_F_ORPHAN macro */
                uint8_t* data;
                union smb_read *io_copy=talloc_memdup_type(NULL, io, union smb_read);

                if (! io_copy)
                        return NT_STATUS_NO_MEMORY;

#warning we are ignoring read_for_execute as far as the cache goes
                io_copy->generic.in.read_for_execute=io->readx.in.read_for_execute;
                io_copy->generic.in.offset=read_position;
                io_copy->generic.in.mincnt=read_block;
                io_copy->generic.in.maxcnt=read_block;
                /* what is generic.in.remaining for? */
                io_copy->generic.in.remaining = MIN(65535,read_remaining);
                io_copy->generic.out.nread=0;

#warning someone must own io_copy, tree, maybe?
                data=talloc_zero_size(io_copy, io_copy->generic.in.maxcnt);
                DEBUG(5,("Talloc read-ahead buffer %p size %d\n",data, io_copy->generic.in.maxcnt));
                if (! data) {
                        talloc_free(io_copy);
                        return NT_STATUS_NO_MEMORY;
                }
                io_copy->generic.out.data=data;

                /* are we able to pull anything from the cache to validate this read-ahead?
                   NOTE: there is no point in reading ahead merely to re-validate the
                   cache if we don't have oplocks and can't save it....
                   ... or maybe there is if we think a read will come that can be matched
                   up to this reponse while it is still on the wire */
#warning so we need to distinguish between pipe-line read-ahead and revalidation
                if (/*(cache->status & CACHE_READ)!=0 && */
                                cache_len(cache) >
                                        (io_copy->generic.in.offset + io_copy->generic.in.mincnt) &&
                                cache->validated_extent <
                                        (io_copy->generic.in.offset + io_copy->generic.in.maxcnt)) {
                        ssize_t pre_fill;

                        pre_fill = cache_raw_read(cache, data,
                                                                          io_copy->generic.in.offset,
                                                                          io_copy->generic.in.maxcnt);
                        DEBUG(5,("Data read into %p %d\n",data, pre_fill));
                        if (pre_fill > 0 && pre_fill >= io_copy->generic.in.mincnt) {
                                io_copy->generic.out.nread=pre_fill;
                                read_block=pre_fill;
                        }
                }

                c_req = proxy_smb_raw_read_send(ntvfs, io_copy, f, NULL);

                if (c_req) {
                        private->readahead_spare--;
                        f->readahead_pending++;
                        DEBUG(CACHE_DEBUG_LEVEL,("Read-ahead level %d request %p offset=%d size=%d\n",io_copy->generic.level,c_req,(int)read_position,(int)read_block));
                        if (cache->readahead_extent < read_position+read_block)
                                cache->readahead_extent=read_position+read_block;
                        ADD_ASYNC_RECV_TAIL(c_req, io_copy, NULL, f, async_read_cache_save, NT_STATUS_INTERNAL_ERROR);
                        /* so we can decrease read-ahead counter for this session */
                        ADD_ASYNC_RECV_TAIL(c_req, io_copy, NULL, f, async_readahead_dec, NT_STATUS_INTERNAL_ERROR);
                        ASYNC_RECV_TAIL_HANDLER_ORPHAN(io_copy, async_read_handler);

                        /* Make these be owned by the async struct so they are freed when the callback ends or is cancelled */
                        talloc_steal(c_req->async.private, c_req);
                        talloc_steal(c_req->async.private, io_copy);
                        read_position+=read_block;
                } else {
                        DEBUG(CACHE_DEBUG_LEVEL,("Read-ahead request FAILED offset=%d size=%d\n",(int)read_position,(int)read_block));
                        talloc_free(io_copy);
                        break;
                }

        }
        DEBUG(CACHE_DEBUG_LEVEL,("DONE: Asking read-aheads\n"));
        return NT_STATUS_OK;
}

struct proxy_validate_parts_parts {
        struct proxy_Read* r;
        struct ntvfs_request *req;
        struct proxy_file *f;
        struct async_read_fragments *fragments;
        off_t offset;
        ssize_t remaining;
        bool complete;
        declare_checksum(digest);
        struct MD5Context context;
};

NTSTATUS proxy_validate_complete(struct proxy_validate_parts_parts *parts);
NTSTATUS async_proxy_validate_parts(struct async_info *async, void* io1, void* io2, NTSTATUS status);
static NTSTATUS proxy_validate_parts(struct ntvfs_module_context *ntvfs,
                                     struct proxy_validate_parts_parts *parts);

/* this will be the new struct proxy_Read based read function, for now
   it just deals with non-cached based validate to a regular server */
static NTSTATUS proxy_validate(struct ntvfs_module_context *ntvfs,
                           struct ntvfs_request *req,
                           struct proxy_Read *r,
                           struct proxy_file *f)
{
        struct proxy_private *private = ntvfs->private_data;
        struct proxy_validate_parts_parts *parts;
        struct async_read_fragments *fragments;
        NTSTATUS status;

        if (!f) return NT_STATUS_INVALID_HANDLE;

        DEBUG(5,("%s: fnum=%d **** %lld bytes \n\n\n\n",__LOCATION__,f->fnum,(long long int)r->in.maxcnt));

        parts = talloc_zero(req, struct proxy_validate_parts_parts);
        DEBUG(5,("%s: parts=%p\n",__FUNCTION__,parts));
        NT_STATUS_HAVE_NO_MEMORY(parts);

        fragments = talloc_zero(parts, struct async_read_fragments);
        NT_STATUS_HAVE_NO_MEMORY(fragments);

        parts->fragments=fragments;

        parts->r=r;
        parts->f=f;
        parts->req=req;
        /* processed offset */
        parts->offset=r->in.offset;
        parts->remaining=r->in.maxcnt;
        fragments->async=true;

        MD5Init (&parts->context);

        /* start a read-loop which will continue in the callback until it is
           all done */
        status=proxy_validate_parts(ntvfs, parts);
        if (parts->complete) {
            /* Make sure we are not async */
            DEBUG(5,("%s: completed EARLY\n",__FUNCTION__));
            return proxy_validate_complete(parts);
        }

        /* Assert if status!=NT_STATUS_OK then parts->complete==true */
        req->async_states->state |= NTVFS_ASYNC_STATE_ASYNC;
        DEBUG(5,("%s: returning ASYNC\n",__FUNCTION__));
        return status;
}

NTSTATUS proxy_validate_complete(struct proxy_validate_parts_parts *parts)
{
        NTSTATUS status;
        struct proxy_Read* r=parts->r;
        struct proxy_file *f=parts->f;

        DEBUG(5,("%s: %d/%d bytes \n\n\n\n",__LOCATION__,r->out.nread,r->in.maxcnt));

        MD5Final(parts->digest, &parts->context);

        status = parts->fragments->status;
        r->out.result = status;
        r->out.response.generic.count=r->out.nread;
        r->out.cache_name.count=0;

        DEBUG(5,("%s: %s nread=%d\n",__FUNCTION__, get_friendly_nt_error_msg (status),
                 r->out.response.generic.count));

        DEBUG(5,("Anticipated validated digest for size: %lld\n", (long long) r->in.maxcnt));
        dump_data (5, r->in.digest.digest, sizeof(parts->digest));
        DEBUG(5,("read digest for size %lld\n",(long long) parts->offset));
        dump_data (5, parts->digest, sizeof(parts->digest));

        if (NT_STATUS_IS_OK(status) &&
            (memcmp(parts->digest, r->in.digest.digest, sizeof(parts->digest))==0)) {
                r->out.flags = PROXY_USE_CACHE | PROXY_VALIDATE;
                DEBUG(5,("======= VALIDATED FINE \n\n\n"));
        } else {
                if (r->in.flags & PROXY_USE_ZLIB) {
                        ssize_t size = r->out.response.generic.count;
                        DEBUG(5,("======= VALIDATED WRONG; compress size %d \n\n\n",size));
                        if (compress_block(r->out.response.generic.data, &size) ) {
                                r->out.flags|=PROXY_USE_ZLIB;
                                r->out.response.compress.count=size;
                                r->out.response.compress.data=r->out.response.generic.data;
                                DEBUG(3,("%s: Compressed from %d to %d = %d%%\n",
                                        __LOCATION__,r->out.nread,size,size*100/r->out.nread));
                        }
                }
                /* return cache filename as a ghastly hack for now */
                r->out.cache_name.s=f->cache->cache_name;
                r->out.cache_name.count=strlen(r->out.cache_name.s)+1;
                DEBUG(5,("%s: writing cache name: %s\n",__LOCATION__, f->cache->cache_name));
                /* todo: what about tiny files, buffer to small, don't validate tiny files <1K */
        }

        /* assert: this must only be true if we are in a callback */
        if (parts->req->async_states->state & NTVFS_ASYNC_STATE_ASYNC) {
                /* we are async complete, we need to call the sendfn */
                parts->req->async_states->status=status;
                DEBUG(5,("Fragments async response sending\n"));

                parts->req->async_states->send_fn(parts->req);
                return NT_STATUS_OK;
        }
        return status;
}

NTSTATUS async_proxy_validate_parts(struct async_info *async, void* io1, void* io2, NTSTATUS status)
{
        struct smbcli_request *c_req = async->c_req;
        struct ntvfs_request *req = async->req;
        struct proxy_file *f = async->f;
        struct ntvfs_module_context *ntvfs = async->proxy->ntvfs;
        struct async_read_fragment* fragment=talloc_get_type_abort(io2, struct async_read_fragment);
        /* this is the io against which the fragment is to be applied */
        struct proxy_validate_parts_parts *parts = talloc_get_type_abort(io1, struct proxy_validate_parts_parts);
        struct proxy_Read* r=parts->r;
        /* this is the io for the read that issued the callback */
        union smb_read *io_frag = fragment->io_frag;
        struct async_read_fragments* fragments=fragment->fragments;

        /* if request is not already received by a chained handler, read it */
        if (c_req) status=smb_raw_read_recv(c_req, io_frag);
        DEBUG(5,("%s: status %s\n",__FUNCTION__,get_friendly_nt_error_msg (status)));
        DEBUG(5,("\n\n%s: parts=%p c_req=%p io_frag=%p read %lld\n",__LOCATION__,parts, c_req, io_frag,(long long int)io_frag->generic.out.nread));

        fragment->status=status;

        if (NT_STATUS_IS_OK(status)) {
                /* TODO: If we are not sequentially "next" the queue until we can do it */
                /* log this data in r->out.generic.data */
                /* Find memcpy window, copy data from the io_frag to the io */

                /* Also write validate to cache */
                if (f && f->cache) {
                        cache_save(f->cache, io_frag->generic.out.data, io_frag->generic.out.nread, io_frag->generic.in.offset);
                }

                /* extent is the last byte we (don't) read for this frag */
                ssize_t extent = io_frag->generic.in.offset + io_frag->generic.out.nread;
                /* start_offset is the file offset we first care about */
                off_t start_offset=MAX(io_frag->generic.in.offset, r->in.offset);
                /* Don't want to go past mincnt cos we don't have the buffer */
                off_t io_extent=r->in.offset + r->in.mincnt;
                off_t end_offset=MIN(io_extent, extent);

                /* ASSERT(start_offset <= end_offset) */
                /* ASSERT(start_offset <= io_extent) */
                /* Don't copy beyond buffer */
                if (! (start_offset >= io_extent)) {
                        uint8_t* dst=r->out.response.generic.data + (start_offset - r->in.offset);
                        uint8_t* src=io_frag->generic.out.data+(start_offset - io_frag->generic.in.offset);
                        /* src == dst in cases where we did not latch onto someone elses
                           read, but are handling our own */
                        if (src != dst)
                                memcpy(dst, src, end_offset - start_offset);
                        r->out.nread=end_offset - r->in.offset;
                        DEBUG(5,("%s: nread %lld ++++++++++++++++++\n", __LOCATION__,(long long int)r->out.nread));
                }

                MD5Update(&parts->context, io_frag->generic.out.data,
                          io_frag->generic.out.nread);

                parts->fragments->status=status;
                status=proxy_validate_parts(ntvfs, parts);
        } else {
                parts->fragments->status=status;
        }

        DLIST_REMOVE(fragments->fragments, fragment);
        /* this will free the io_frag too */
        talloc_free(fragment);

        if (parts->complete || NT_STATUS_IS_ERR(status)) {
                /* this will call sendfn, the chain handler won't know... but
                   should have no more handlers queued */
                return proxy_validate_complete(parts);
        }

        return NT_STATUS_OK;
}

/* continue a read loop, possibly from a callback */
static NTSTATUS proxy_validate_parts(struct ntvfs_module_context *ntvfs,
                                     struct proxy_validate_parts_parts *parts)
{
        struct proxy_private *private = ntvfs->private_data;
        union smb_read *io_frag;
        struct async_read_fragment *fragment;
        struct smbcli_request *c_req = NULL;
        ssize_t size=private->tree->session->transport->negotiate.max_xmit \
                                                - (MIN_SMB_SIZE+32);

        /* Have we already read enough? */
        if (parts->offset >= (parts->r->in.offset + parts->r->in.maxcnt)) {
            parts->complete=true;
            return NT_STATUS_OK;
        }

        size=MIN(size, parts->remaining);

        fragment=talloc_zero(parts->fragments, struct async_read_fragment);
        NT_STATUS_HAVE_NO_MEMORY(fragment);

        io_frag = talloc_zero(fragment, union smb_read);
        NT_STATUS_HAVE_NO_MEMORY(io_frag);

        io_frag->generic.out.data = talloc_size(io_frag, size);
        NT_STATUS_HAVE_NO_MEMORY(io_frag->generic.out.data);

        io_frag->generic.level = RAW_READ_GENERIC;
        io_frag->generic.in.file.fnum = parts->r->in.fnum;
        io_frag->generic.in.offset = parts->offset;
        io_frag->generic.in.mincnt = size;
        io_frag->generic.in.maxcnt = size;
        io_frag->generic.in.remaining = 0;
#warning maybe true is more permissive?
        io_frag->generic.in.read_for_execute = false;

        DEBUG(5,("%s: issue part read offset=%lld, size=%lld,%lld\n",__LOCATION__,
                 (long long int)io_frag->generic.in.offset,
                 (long long int)io_frag->generic.in.mincnt,
                 (long long int)io_frag->generic.in.maxcnt));

        //c_req = smb_raw_read_send(ntvfs, io_frag, parts->f, parts->r);
        c_req = smb_raw_read_send(private->tree, io_frag);
        NT_STATUS_HAVE_NO_MEMORY(c_req);

        parts->offset+=size;
        parts->remaining-=size;
        fragment->c_req = c_req;
        fragment->io_frag = io_frag;
        fragment->fragments=parts->fragments;
        DLIST_ADD(parts->fragments->fragments, fragment);

        { void* req=NULL;
        ADD_ASYNC_RECV_TAIL(c_req, parts, fragment, parts->f, async_proxy_validate_parts, NT_STATUS_INTERNAL_ERROR);
        ASYNC_RECV_TAIL_F_ORPHAN(io_frag, async_read_handler, parts->f, c_req->async.private, NT_STATUS_UNSUCCESSFUL);
        }

        DEBUG(5,("%s: issued read parts=%p c_req=%p io_frag=%p\n",__LOCATION__,parts, c_req, io_frag));

        return NT_STATUS_OK;
}

/*
  read from a file
*/
static NTSTATUS proxy_read(struct ntvfs_module_context *ntvfs,
                          struct ntvfs_request *req, union smb_read *io)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;
        struct proxy_file *f;
        struct async_read_fragments *fragments=NULL;
        /* how much of read-from-cache is certainly valid */
        ssize_t valid=0;
        off_t offset=io->generic.in.offset+valid;
        off_t limit=io->generic.in.offset+io->generic.in.mincnt;

        SETUP_PID;

        if (io->generic.level != RAW_READ_GENERIC &&
            private->map_generic) {
                return ntvfs_map_read(ntvfs, req, io);
        }

        SETUP_FILE_HERE(f);

        DEBUG(3,("\n%s() fnum=%d offset=%lld, mincnt=%d, maxcnt=%d\n",__FUNCTION__,
                  io->generic.in.file.fnum,
                  io->generic.in.offset,
                  io->generic.in.mincnt,
                  io->generic.in.maxcnt));

        io->generic.out.nread=0;

        /* if we have oplocks and know the files size, don't even ask the server
           for more */
        if (f->oplock) {
            if (io->generic.in.offset >= f->metadata->info_data.size) {
                io->generic.in.mincnt=0;
                io->generic.in.maxcnt=0;
                io->generic.out.nread=0;
                DEBUG(5,("Reading beyond known length %lld; return 0\n",(long long)f->metadata->info_data.size));
                return NT_STATUS_OK;
            } else {
                io->generic.in.mincnt=MIN(io->generic.in.mincnt,
                                          f->metadata->info_data.size - io->generic.in.offset);
                io->generic.in.maxcnt=MIN(io->generic.in.maxcnt,
                                          f->metadata->info_data.size - io->generic.in.offset);
            }
            DEBUG(5,("Oplock and known size, limiting read to %lld (s=%d)\n",
                     f->metadata->info_data.size, io->generic.in.mincnt));
        }


        /* attempt to read from cache.  if nread becomes non-zero then we
           have cache to validate. Instead of returning "valid" value, cache_read
           should probably return an async_read_fragment structure */

        if (private->cache_enabled) {
                NTSTATUS status=cache_smb_raw_read(f->cache, ntvfs, req, io, &valid);

                if (NT_STATUS_IS_OK(status)) {
                        /* if we read enough valid data, return it */
                        if (valid > 0 && valid>=io->generic.in.mincnt) {
                                /* valid will not be bigger than maxcnt */
                                io->generic.out.nread=valid;
                                DEBUG(1,("Read from cache offset=%d size=%d\n",
                                                 (int)(io->generic.in.offset),
                                                 (int)(io->generic.out.nread)) );
                                return status;
                        }
                }
                DEBUG(5,("Cache read status: %s\n",get_friendly_nt_error_msg (status)));
        }

        fragments=talloc_zero(req, struct async_read_fragments);
        fragments->async=!!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC);
        /* See if there are pending reads that would satisfy this request
           We have a validated read up to io->generic.out.nread. Anything between
           this and mincnt MUST be read, but we could first try and attach to
           any pending read-ahead on the same file.
           If those read-aheads fail we will re-issue a regular read from the
           callback handler and hope it hasn't taken too long. */

        /* offset is the extentof the file from which we still need to find
           matching read-requests. */
        offset=io->generic.in.offset+valid;
        /* limit is the byte beyond the last byte for which we need a request.
           This used to be mincnt, but is now maxcnt to cope with validate reads.
           Maybe we can switch back to mincnt when proxy_read struct is used
           instead of smb_read.
        */
        limit=io->generic.in.offset+io->generic.in.maxcnt;

        while (offset < limit) {
                /* Should look for the read-ahead with offset <= in.offset+out.nread
                   with the longest span, but there is only likely to be one anyway so
                   just take the first */
                struct async_info* pending=private->pending;
                union smb_read *readahead_io=NULL;
                DEBUG(5,("Looping reads from offset=%lld, end=%lld\n",offset,limit));
                while(pending) {
                        if (pending->c_req->async.fn == async_read_handler) {
                                struct async_info *async=talloc_get_type_abort(pending->c_req->async.private, struct async_info);
                                readahead_io=talloc_get_type_abort(async->parms, union smb_read);

                                if (readahead_io->generic.in.file.fnum == io->generic.in.file.fnum &&
                                        readahead_io->generic.in.offset <= offset &&
                                        readahead_io->generic.in.offset +
                                                readahead_io->generic.in.mincnt > offset) break;
                        }
                        readahead_io=NULL;
                        pending=pending->next;
                }
                /* ASSERT(readahead_io == pending->c_req->async.params) */
                if (pending && readahead_io) {
                        struct async_read_fragment *fragment=talloc_zero(req, struct async_read_fragment);
                        fragment->fragments=fragments;
                        fragment->io_frag=readahead_io;
                        fragment->c_req = pending->c_req;
                        /* we found one, so attach to it. We DO need a talloc_reference
                           because the original send_fn might be called before ALL chained
                           handlers, and our handler will call its own send_fn first. ugh.
                           Maybe we need to seperate reverse-mapping callbacks with data users? */
                        /* Note: the read-ahead io is passed as io, and our req io is
                           in io_frag->io */
                        //talloc_reference(req, pending->req);
                        DEBUG(5,("Attach to read for offset=%lld length=%d\n",
                                         readahead_io->generic.in.offset,
                                         readahead_io->generic.in.mincnt));
                        ADD_ASYNC_RECV_TAIL(pending->c_req, io, fragment, f,
                                                                async_read_fragment, NT_STATUS_INTERNAL_ERROR);
                        DEBUG(5,("Attached OK\n"));
#warning we don't want to return if we fail to attach, just break
                        DLIST_ADD(fragments->fragments, fragment);
                        /* updated offset for which we have reads */
                        offset=readahead_io->generic.in.offset + readahead_io->generic.in.mincnt;
                } else {
                        /* there are no pending reads to fill this so issue one up to
                           the maximum supported read size. We could see when the next
                           pending read is (if any) and only read up till there... later...
                           Issue a fragment request for what is left, clone io.
                           In the case that there were no fragments this will be the orginal read
                           but with a cloned io struct */
                        off_t next_offset;
                        struct proxy_Read *r=NULL; /* used only for VALIDATE promotion */
                        struct async_read_fragment *fragment=talloc_zero(req, struct async_read_fragment);
                        union smb_read *io_frag=talloc_memdup_type(req, io, union smb_read);
                        ssize_t offset_inc=offset-io_frag->generic.in.offset;
                        /* 250 is a guess at ndr rpc overheads */
                        ssize_t readsize=MIN(PROXY_NTIOCTL_MAXDATA,
                                                private->tree->session->transport->negotiate.max_xmit) \
                                                - (MIN_SMB_SIZE+32);
                        if (readsize > 0xFFFF) readsize = 0xFFFF; /* - (MIN_SMB_SIZE+250) ?? */
                        readsize=MIN(limit-offset, readsize);

                        DEBUG(5,("Issuing direct read\n"));
                        /* reduce the cached read (if any). nread is unsigned */
                        if (io_frag->generic.out.nread > offset_inc) {
                                io_frag->generic.out.nread-=offset_inc;
                                /* don't make nread buffer look too big */
                                if (io_frag->generic.out.nread > readsize)
                                        io_frag->generic.out.nread = readsize;
                        } else {
                                io_frag->generic.out.nread=0;
                        }
                        /* adjust the data pointer so we read to the right place */
                        io_frag->generic.out.data+=offset_inc;
                        io_frag->generic.in.offset=offset;
                        io_frag->generic.in.maxcnt=readsize;
                        /* we don't mind mincnt being smaller if this is the last frag,
                           but then we can already handle it being bigger but not reached...
                           The spell would be:
                            MIN(io_frag->generic.in.mincnt, io_frag->generic.in.maxcnt);
                        */
                        io_frag->generic.in.mincnt=readsize;
                        fragment->fragments=fragments;
                        fragment->io_frag=io_frag;
#warning attach to send_fn handler
                        /* what if someone attaches to us? Our send_fn is called from our
                        chained handler which will be before their handler and io will
                        already be freed. We need to keep a reference to the io and the data
                        but we don't know where it came from in order to take a reference.
                        We need therefore to tackle calling of send_fn AFTER all other handlers */

                        /* Calculate next offset (in advance) */
                        next_offset=io_frag->generic.in.offset + io_frag->generic.in.mincnt;

                        /* if we are (going to be) the last fragment and we are in VALIDATE
                           mode, see if we can do a bulk validate now.
                           io->generic.in.mincnt == io->generic.in.maxcnt is to make sure we
                           don't do a validate on a receive validate read
                         */
                        if (private->cache_validatesize && PROXY_REMOTE_SERVER(private) &&
                                next_offset >= limit && (f->cache && f->cache->status & CACHE_VALIDATE)) {
                                ssize_t length=private->cache_validatesize;
                                declare_checksum(digest);

                                DEBUG(5,("last read, maybe mega validate: frag length %zu, offset %llu\n",
                                                 length, (unsigned long long) offset));
                                NTSTATUS status=cache_smb_raw_checksum(f->cache, offset, &length, digest);
                                /* no point in doing it if md5'd length < current out.nread
                                   remember: out.data contains this requests cached response
                                   if validate succeeds */
                                if (NT_STATUS_IS_OK(status) && (length > io_frag->generic.out.nread)) {
                                        /* upgrade the read, allocate the proxy_read struct here
                                           and fill in the extras, no more out-of-band stuff */
                                        DEBUG(5,("%s: Promoting to validate read: %lld\n",__FUNCTION__,(long long) length));
                        dump_data (5, digest, sizeof(digest));

                                        r=talloc_zero(io_frag, struct proxy_Read);
                                        memcpy(r->in.digest.digest, digest, sizeof(digest));
                                        r->in.flags |= PROXY_VALIDATE | PROXY_USE_CACHE;
                                        io_frag->generic.in.maxcnt = length;
                                        r->in.mincnt=io_frag->generic.in.mincnt;
                                        /* the proxy send function will calculate the checksum based on *data */
                                } else {
                                        /* try bulk read */
                                        if (f->oplock) {
                                                DEBUG(5,("%s: *** faking bulkd read\n\n",__LOCATION__));
                                                r=talloc_zero(io_frag, struct proxy_Read);
                                                r->in.flags |= PROXY_VALIDATE | PROXY_USE_CACHE;//| PROXY_USE_ZLIB;
                                                io_frag->generic.in.maxcnt = MIN(f->metadata->info_data.size, private->cache_validatesize);
                                                r->in.mincnt=io_frag->generic.in.maxcnt;
                                                r->in.mincnt=io_frag->generic.in.mincnt;
                                        }
                                        /* not enough in cache to make it worthwhile anymore */
                                        DEBUG(5,("VALIDATE DOWNGRADE 1, no more on this file: frag length %zu, offset %llu, cache=%x len=%lld\n",
                                                         length, (unsigned long long) offset, (f->cache)?(f->cache->status):0,
                                                         (unsigned long long)length));
                                        //cache_handle_novalidate(f);
                                        DEBUG(5,("VALIDATE DOWNGRADE 1, no more on this file: frag length %zu, offset %llu, cache=%x\n",
                                                         length, (unsigned long long) offset, (f->cache)?(f->cache->status):0));
                                }
                        } else {
                                if (f->cache && f->cache->status & CACHE_VALIDATE) {
                                        DEBUG(5,(">>>Not last frag, no validate read: %lld %lld\n",
                                                         (long long) next_offset,
                                                         (long long) limit));
                                }
                        }

                        DEBUG(5,("Frag read sending offset=%lld min=%d, size=%d\n",
                                         io_frag->generic.in.offset,io_frag->generic.in.mincnt,
                                         io_frag->generic.in.maxcnt));
                        c_req = proxy_smb_raw_read_send(ntvfs, io_frag, f, r);
                        DEBUG(5,("Frag read sent offset=%lld size=%d MID=%d\n",
                                         io_frag->generic.in.offset,io_frag->generic.in.maxcnt,c_req->mid));
                        fragment->c_req=c_req;
                        DLIST_ADD(fragments->fragments, fragment);
                        ADD_ASYNC_RECV_TAIL(c_req, io_frag, NULL, f, async_read_cache_save, NT_STATUS_INTERNAL_ERROR);
                        ADD_ASYNC_RECV_TAIL(c_req, io, fragment, f, async_read_fragment, NT_STATUS_INTERNAL_ERROR);
                        DEBUG(5,("Frag response chained\n"));
                        /* normally we would only install the chain_handler if we wanted async
                        response, but as it is the async_read_fragment handler that calls send_fn
                        based on fragments->async, instead of async_chain_handler, we don't
                        need to worry about this call completing async'ly while we are
                                waiting on the other attached calls. Otherwise we would not attach
                        the async_chain_handler (via async_read_handler) because of the wait
                        below */
                        { /* We don't want the chain handler calling send_fn as it is done by the fragment handler */
                                void* req=NULL;
                                /* call async_chain_hander not read handler so that folk can't
                                   attach to it, till we solve the problem above */
                                ASYNC_RECV_TAIL_HANDLER_ORPHAN(io, async_chain_handler);
                        }
                        offset = next_offset;
                }
                DEBUG(5,("Next fragment\n"));
        }

        /* do we still need a final fragment? Issue a read */

        DEBUG(5,("No frags left to read\n"));


        /* issue new round of read-aheads */
        DEBUG(5,("== Read aheads asread-%d\n",io->generic.in.mincnt));
        if (f->cache && ! (f->cache->status & CACHE_VALIDATE)) read_ahead(f, ntvfs, io, io->generic.in.mincnt);
        DEBUG(5,("== Done Read aheads\n"));

        /* If we have fragments but we are not called async, we must sync-wait on them */
        /* did we map the entire request to pending reads? */
        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                struct async_read_fragment *fragment;
                DEBUG(5,("Sync waiting\n"));
                /* fragment get's free'd during the chain_handler so we start at
                   the top each time */
                for (fragment = fragments->fragments; fragment; fragment = fragments->fragments) {
                        /* Any fragments async handled while we sync-wait on one
                           will remove themselves from the list and not get sync waited */
                        sync_chain_handler(fragment->c_req);
                        /* if we have a non-ok result AND we know we have all the responses
                           up to extent, then we could quit the loop early and change the
                           fragments->async to true so the final irrelevant responses would
                           come async and we could send our response now - but we don't
                           track that detail until we have cache-maps that we can use to
                           track the responded fragments and combine responsed linear extents
                        if (! NT_STATUS_IS_OK(fragments->status) && xxx ) */
                }
                DEBUG(5,("Sync return of proxy_read: %s\n",get_friendly_nt_error_msg (fragments->status)));
                return fragments->status;
        }

        DEBUG(5,("Async returning\n"));
        req->async_states->state |= NTVFS_ASYNC_STATE_ASYNC;
        return NT_STATUS_OK;
        }

/*
  a handler to de-fragment async write replies back to one request.
  Can cope with out-of-order async responses by waiting for all responses
  on an NT_STATUS_OK case so that nwritten is properly adjusted
 */
NTSTATUS async_write_fragment(struct async_info *async, void* io1, void* io2, NTSTATUS status)
{
        struct smbcli_request *c_req = async->c_req;
        struct ntvfs_request *req = async->req;
        struct proxy_file *f=async->f;
        struct async_write_fragment* fragment=talloc_get_type_abort(io2, struct async_write_fragment);
        /* this is the io against which the fragment is to be applied */
        union smb_write *io = talloc_get_type_abort(io1, union smb_write);
        /* this is the io for the write that issued the callback */
        union smb_write *io_frag = fragment->io_frag; /* async->parms; */
        struct async_write_fragments* fragments=fragment->fragments;
        ssize_t extent=0;

        /* if request is not already received by a chained handler, read it */
#warning the queuer of the request should first push a suitable decoder, they should not scatter handlers generically
        if (c_req) status=smb_raw_write_recv(c_req, io_frag);

        DEBUG(3,("%s async_write status: %s\n",__FUNCTION__,
                         get_friendly_nt_error_msg(status)));

        fragment->status = status;

        DLIST_REMOVE(fragments->fragments, fragment);

        /* did this one fail? */
        if (! NT_STATUS_IS_OK(fragment->status)) {
                if (NT_STATUS_IS_OK(fragments->status)) {
                        fragments->status=fragment->status;
                }
        } else {
        /* No fragments have yet failed, keep collecting responses */
        extent = io_frag->generic.in.offset + io_frag->generic.out.nwritten;

        /* we broke up the write so it could all be written. If only some has
           been written of this block, and then some of then next block,
           it could leave unwritten holes! We will only acknowledge up to the
           first partial write, and let the client deal with it.
           If server can return NT_STATUS_OK for a partial write so can we */
        if (io_frag->generic.out.nwritten != io_frag->generic.in.count) {
                DEBUG(4,("Fragmented write only partially successful\n"));

                /* Shrink the master nwritten */
                if ( ! fragments->partial ||
                                (io->generic.in.offset + io->generic.out.nwritten) > extent)  {
                        io->generic.out.nwritten = extent - io->generic.in.offset;
                }
                /* stop any further successes from extended the partial write */
                fragments->partial=true;
        } else {
                /* only grow the master nwritten if we haven't logged a partial write */
                if (! fragments->partial &&
                                (io->generic.in.offset + io->generic.out.nwritten) < extent ) {
                        io->generic.out.nwritten = extent - io->generic.in.offset;
                }
        }
        }

        /* if this was the last fragment, clean up */
        if (! fragments->fragments) {
                DEBUG(5,("Async write re-fragmented with %d of %d\n",
                                 io->generic.out.nwritten,
                                 io->generic.in.count));
                if (NT_STATUS_IS_OK(fragments->status)) {
                        cache_handle_save(f, io->generic.in.data, io->generic.out.nwritten,
                                                          io->generic.in.offset);
                        if (f->metadata->info_data.size < io->generic.in.offset+io->generic.in.count) {
                                f->metadata->info_data.size=io->generic.in.offset+io->generic.in.count;
                        }
                }
                if (fragments->async) {
                        req->async_states->status=fragments->status;
#warning its not good freeing early if other pending requests have io allocated against this request which will now be freed
                        req->async_states->send_fn(req);
                        DEBUG(5,("Async response sent\n"));
                } else {
                        DEBUG(5,("Fragments SYNC return\n"));
                }
        }

        return status;
}

/*
  a handler for async write replies
 */
NTSTATUS async_write_cache_save(struct async_info *async, void* io1, void* io2, NTSTATUS status)
{
        struct smbcli_request *c_req = async->c_req;
        struct ntvfs_request *req = async->req;
        struct proxy_file *f=async->f;
        union smb_write *io=async->parms;

        if (c_req)
                status = smb_raw_write_recv(c_req, async->parms);

        cache_handle_save(f, io->generic.in.data,
                io->generic.out.nwritten,
                io->generic.in.offset);

        return status;
}

/*
  write to a file
*/
static NTSTATUS proxy_write(struct ntvfs_module_context *ntvfs,
                           struct ntvfs_request *req, union smb_write *io)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;
        struct proxy_file *f;

        SETUP_PID;

        if (io->generic.level != RAW_WRITE_GENERIC &&
            private->map_generic) {
                return ntvfs_map_write(ntvfs, req, io);
        }
        SETUP_FILE_HERE(f);

        DEBUG(5,("proxy_write offset=%lld size=%d\n",io->generic.in.offset, io->generic.in.count));
#warning ERROR get rid of this
        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                NTSTATUS status;
                if (PROXY_REMOTE_SERVER(private)) {
                        /* Do a proxy write */
                        status=proxy_smb_raw_write(ntvfs, io, f);
                } else if (io->generic.in.count >
                        private->tree->session->transport->negotiate.max_xmit) {

                        /* smbcli_write can deal with large writes, which are bigger than
                           tree->session->transport->negotiate.max_xmit */
                        ssize_t size=smbcli_write(private->tree,
                                                io->generic.in.file.fnum,
                                                io->generic.in.wmode,
                                                io->generic.in.data,
                                                io->generic.in.offset,
                                                io->generic.in.count);

                        if (size==io->generic.in.count || size > 0) {
                                io->generic.out.nwritten=size;
                                status=NT_STATUS_OK;
                        } else {
                                status=NT_STATUS_UNSUCCESSFUL;
                        }
                } else {
                        status=smb_raw_write(private->tree, io);
                }

                /* Save write in cache */
                if (NT_STATUS_IS_OK(status)) {
                        cache_handle_save(f, io->generic.in.data,
                                io->generic.out.nwritten,
                                io->generic.in.offset);
                        if (f->metadata->info_data.size <
                            io->generic.in.offset+io->generic.in.count) {
                                f->metadata->info_data.size=io->generic.in.offset+io->generic.in.count;
                        }
                }

                return status;
        }

        /* smb_raw_write_send can't deal with large writes, which are bigger than
           tree->session->transport->negotiate.max_xmit so we have to break it up
           trying to preserve the async nature of the call as much as possible */
        if (PROXY_REMOTE_SERVER(private)) {
                DEBUG(5,("== %s call proxy_smb_raw_write_send\n",__FUNCTION__));
                c_req = proxy_smb_raw_write_send(ntvfs, io, f);
                ADD_ASYNC_RECV_TAIL(c_req, io, NULL, f, async_write_cache_save, NT_STATUS_INTERNAL_ERROR);
        } else if (io->generic.in.count <=
                        private->tree->session->transport->negotiate.max_xmit) {
                DEBUG(5,("== %s call smb_raw_write_send\n",__FUNCTION__));
        c_req = smb_raw_write_send(private->tree, io);
                ADD_ASYNC_RECV_TAIL(c_req, io, NULL, f, async_write_cache_save, NT_STATUS_INTERNAL_ERROR);
        } else {
                ssize_t remaining = io->generic.in.count;
#warning Need an audit of these magin numbers MIN_SMB_SIZE+32
                int block = (private->tree->session->transport->negotiate.max_xmit - (MIN_SMB_SIZE+32));
                int done = 0;
                struct async_write_fragments *fragments = talloc_zero(req, struct async_write_fragments);

                DEBUG(3,("== %s Client sending too-big write sized %d, negotiated limit %d\n",
                                 __FUNCTION__, io->generic.in.count,
                                 private->tree->session->transport->negotiate.max_xmit));

                fragments->io = io;
                io->generic.out.nwritten=0;
                io->generic.out.remaining=0;

                do {
                        union smb_write *io_frag = talloc_zero(fragments, union smb_write);
                        struct async_write_fragment *fragment = talloc_zero(fragments, struct async_write_fragment);
                        ssize_t size = MIN(block, remaining);

                        fragment->fragments = fragments;
                        fragment->io_frag = io_frag;

                        io_frag->generic.level                  = io->generic.level;
                        io_frag->generic.in.file.fnum   = io->generic.in.file.fnum;
                        io_frag->generic.in.wmode               = io->generic.in.wmode;
                        io_frag->generic.in.count               = size;
                        io_frag->generic.in.offset              = io->generic.in.offset + done;
                        io_frag->generic.in.data                = io->generic.in.data + done;

                        c_req = proxy_smb_raw_write_send(ntvfs, io_frag, f);
                        if (! c_req) {
                                /* let pending requests clean-up when ready */
                                fragments->status=NT_STATUS_UNSUCCESSFUL;
                                talloc_steal(NULL, fragments);
                                DEBUG(3,("Can't send request fragment\n"));
                                return NT_STATUS_UNSUCCESSFUL;
                        }

                        DEBUG(5,("Frag write sent offset=%lld size=%d MID=%d\n",
                                 io_frag->generic.in.offset,io_frag->generic.in.count,c_req->mid));
                        fragment->c_req=c_req;
                        DLIST_ADD(fragments->fragments, fragment);

//                      ADD_ASYNC_RECV_TAIL(c_req, io_frag, NULL, f, async_write_cache_save, NT_STATUS_INTERNAL_ERROR);
                        ADD_ASYNC_RECV_TAIL(c_req, io, fragment, f, async_write_fragment, NT_STATUS_INTERNAL_ERROR);
                        DEBUG(5,("Frag response chained\n"));

                        remaining       -= size;
                        done            += size;
                } while(remaining > 0);

                /* this strategy has the callback chain attached to each c_req, so we
                   don't use the ASYNC_RECV_TAIL* to install a general one */
        }

        ASYNC_RECV_TAIL_HANDLER(io, async_chain_handler);
}

/*
  a handler for async seek replies
 */
static void async_seek(struct smbcli_request *c_req)
{
        struct async_info *async = c_req->async.private;
        struct ntvfs_request *req = async->req;
        req->async_states->status = smb_raw_seek_recv(c_req, async->parms);
        talloc_free(async);
        req->async_states->send_fn(req);
}

/*
  seek in a file
*/
static NTSTATUS proxy_seek(struct ntvfs_module_context *ntvfs,
                          struct ntvfs_request *req,
                          union smb_seek *io)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;

        SETUP_PID_AND_FILE;

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return smb_raw_seek(private->tree, io);
        }

        c_req = smb_raw_seek_send(private->tree, io);

        ASYNC_RECV_TAIL(io, async_seek);
}

/*
  flush a file
*/
static NTSTATUS proxy_flush(struct ntvfs_module_context *ntvfs,
                           struct ntvfs_request *req,
                           union smb_flush *io)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;

        SETUP_PID;
        switch (io->generic.level) {
        case RAW_FLUSH_FLUSH:
                SETUP_FILE;
                break;
        case RAW_FLUSH_ALL:
                io->generic.in.file.fnum = 0xFFFF;
                break;
        case RAW_FLUSH_SMB2:
                return NT_STATUS_INVALID_LEVEL;
        }

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return smb_raw_flush(private->tree, io);
        }

        c_req = smb_raw_flush_send(private->tree, io);

        SIMPLE_ASYNC_TAIL;
}

/*
  close a file
*/
static NTSTATUS proxy_close(struct ntvfs_module_context *ntvfs,
                           struct ntvfs_request *req, union smb_close *io)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;
        struct proxy_file *f;
        union smb_close io2;
        bool can_clone;

        SETUP_PID;

        if (io->generic.level != RAW_CLOSE_GENERIC &&
            private->map_generic) {
                return ntvfs_map_close(ntvfs, req, io);
        }
        SETUP_FILE_HERE(f);
        /* we free the backend data before we use this value, so save it */
        can_clone=f->can_clone;
        /* Note, we aren't free-ing f, or it's h here. Should we?
           even if file-close fails, we'll remove it from the list,
           what else would we do? Maybe we should not remove until
           after the proxied call completes? */
        DLIST_REMOVE(private->files, f);

        /* Don't send the close on cloned handles unless we are the last one */
        if (f->metadata && --(f->metadata->count)) {
                DEBUG(5,("%s: Fake close of %d, %d left\n",__FUNCTION__,f->fnum, f->metadata->count));
                return NT_STATUS_OK;
        }
        DEBUG(5,("%s: Real close of %d\n",__FUNCTION__, f->fnum));
        /* only close the cache if we aren't keeping references */
        //cache_close(f->cache);

        /* possibly samba can't do RAW_CLOSE_SEND yet */
        if (! (c_req = smb_raw_close_send(private->tree, io))) {
                if (io->generic.level == RAW_CLOSE_GENERIC) {
                        ZERO_STRUCT(io2);
                        io2.close.level = RAW_CLOSE_CLOSE;
                        io2.close.in.file = io->generic.in.file;
                        io2.close.in.write_time = io->generic.in.write_time;
                        io = &io2;
                }
                c_req = smb_raw_close_send(private->tree, io);
                /* destroy handle */
                ntvfs_handle_remove_backend_data(f->h, ntvfs);
        }

        /* If it is read-only, don't bother waiting for the result */
        if (can_clone) {
        DEBUG(5,("%s: not waiting for close response fnum=%d\n",__FUNCTION__,f->fnum));
            return NT_STATUS_OK;
        }

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return smbcli_request_simple_recv(c_req);
        }
DEBUG(0,("%s\n",__LOCATION__));
        SIMPLE_ASYNC_TAIL;
}

/*
  exit - closing files open by the pid
*/
static NTSTATUS proxy_exit(struct ntvfs_module_context *ntvfs,
                          struct ntvfs_request *req)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;

        SETUP_PID;

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return smb_raw_exit(private->tree->session);
        }

        c_req = smb_raw_exit_send(private->tree->session);

        SIMPLE_ASYNC_TAIL;
}

/*
  logoff - closing files open by the user
*/
static NTSTATUS proxy_logoff(struct ntvfs_module_context *ntvfs,
                            struct ntvfs_request *req)
{
        /* we can't do this right in the proxy backend .... */
        return NT_STATUS_OK;
}

/*
  setup for an async call - nothing to do yet
*/
static NTSTATUS proxy_async_setup(struct ntvfs_module_context *ntvfs,
                                 struct ntvfs_request *req,
                                 void *private)
{
        return NT_STATUS_OK;
}

/*
  cancel an async call
*/
static NTSTATUS proxy_cancel(struct ntvfs_module_context *ntvfs,
                            struct ntvfs_request *req)
{
        struct proxy_private *private = ntvfs->private_data;
        struct async_info *a;

        /* find the matching request */
        for (a=private->pending;a;a=a->next) {
                if (a->req == req) {
                        break;
                }
        }

        if (a == NULL) {
                return NT_STATUS_INVALID_PARAMETER;
        }

        return smb_raw_ntcancel(a->c_req);
}

/*
  lock a byte range
*/
static NTSTATUS proxy_lock(struct ntvfs_module_context *ntvfs,
                          struct ntvfs_request *req, union smb_lock *io)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;

        SETUP_PID;

        if (io->generic.level != RAW_LOCK_GENERIC &&
            private->map_generic) {
                return ntvfs_map_lock(ntvfs, req, io);
        }
        SETUP_FILE;

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return smb_raw_lock(private->tree, io);
        }

        c_req = smb_raw_lock_send(private->tree, io);
        SIMPLE_ASYNC_TAIL;
}

/*
  set info on a open file
*/
static NTSTATUS proxy_setfileinfo(struct ntvfs_module_context *ntvfs,
                                 struct ntvfs_request *req,
                                 union smb_setfileinfo *io)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;

        SETUP_PID_AND_FILE;

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return smb_raw_setfileinfo(private->tree, io);
        }
        c_req = smb_raw_setfileinfo_send(private->tree, io);

        SIMPLE_ASYNC_TAIL;
}


/*
  a handler for async fsinfo replies
 */
static void async_fsinfo(struct smbcli_request *c_req)
{
        struct async_info *async = c_req->async.private;
        struct ntvfs_request *req = async->req;
        union smb_fsinfo *fs = async->parms;
        struct proxy_private *private = async->proxy;

        req->async_states->status = smb_raw_fsinfo_recv(c_req, req, fs);

        if (NT_STATUS_IS_OK(req->async_states->status) && (fs->generic.level == RAW_QFS_ATTRIBUTE_INFORMATION ||
                                                           fs->generic.level == RAW_QFS_ATTRIBUTE_INFO)) {
                if (! private->fs_attribute_info && (private->fs_attribute_info=talloc_zero(private, struct fs_attribute_info))) {
                        DEBUG(5,("%s: caching fs_attribute_info\n",__LOCATION__));
                        private->fs_attribute_info->fs_attr=fs->attribute_info.out.fs_attr;
                        private->fs_attribute_info->max_file_component_length=fs->attribute_info.out.max_file_component_length;
                        private->fs_attribute_info->fs_type=talloc_smb_wire_string_dup(private, &(fs->attribute_info.out.fs_type));
                }
        }

        talloc_free(async);
        req->async_states->send_fn(req);
}

/*
  return filesystem space info
*/
static NTSTATUS proxy_fsinfo(struct ntvfs_module_context *ntvfs,
                            struct ntvfs_request *req, union smb_fsinfo *fs)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;

        SETUP_PID;

        DEBUG(5,("%s: level %x\n",__LOCATION__,fs->generic.level));
        /* this value is easy to cache */
        if ((fs->generic.level == RAW_QFS_ATTRIBUTE_INFORMATION ||
             fs->generic.level == RAW_QFS_ATTRIBUTE_INFO) &&
            private->fs_attribute_info) {
                DEBUG(5,("%s: using cached fsinfo\n",__LOCATION__));
                fs->attribute_info.out.fs_attr=private->fs_attribute_info->fs_attr;
                fs->attribute_info.out.max_file_component_length=private->fs_attribute_info->max_file_component_length;
                fs->attribute_info.out.fs_type=talloc_smb_wire_string_dup(req, &(private->fs_attribute_info->fs_type));
                return NT_STATUS_OK;
        }

        /* QFS Proxy */
        if (fs->generic.level == RAW_QFS_PROXY_INFO) {
                fs->proxy_info.out.major_version=1;
                fs->proxy_info.out.minor_version=0;
                fs->proxy_info.out.capability=0;
                return NT_STATUS_OK;
        }

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                NTSTATUS status = smb_raw_fsinfo(private->tree, req, fs);
                if (NT_STATUS_IS_OK(status) && (fs->generic.level == RAW_QFS_ATTRIBUTE_INFORMATION ||
                                                fs->generic.level == RAW_QFS_ATTRIBUTE_INFO)) {
                        if (! private->fs_attribute_info && (private->fs_attribute_info=talloc_zero(private, struct fs_attribute_info))) {
                                DEBUG(5,("%s: caching fs_attribute_info\n",__LOCATION__));
                                private->fs_attribute_info->fs_attr=fs->attribute_info.out.fs_attr;
                                private->fs_attribute_info->max_file_component_length=fs->attribute_info.out.max_file_component_length;
                                private->fs_attribute_info->fs_type=talloc_smb_wire_string_dup(private, &(fs->attribute_info.out.fs_type));
                        }
                }
                return status;
        }
        c_req = smb_raw_fsinfo_send(private->tree, req, fs);

        ASYNC_RECV_TAIL(fs, async_fsinfo);
}

/*
  return print queue info
*/
static NTSTATUS proxy_lpq(struct ntvfs_module_context *ntvfs,
                         struct ntvfs_request *req, union smb_lpq *lpq)
{
        return NT_STATUS_NOT_SUPPORTED;
}

/*
   find_first / find_next caching.
   For now, cache based on directory,search_attributes,search_pattern,ea stuff
   Consider in response:
    * search id
    * search count
    * end of search
    * ea stuff
*/

static union smb_search_data *smb_search_data_dup(void* mem_ctx, const union smb_search_data *file, enum smb_search_data_level data_level) {
        union smb_search_data *result;
        struct smb_wire_string *name;

        result=talloc_zero(mem_ctx, union smb_search_data);
        if (! result) {
                return result;
        }

        *result = *file;

        switch(data_level) {
        case RAW_SEARCH_DATA_SEARCH:
                        if (! (result->search.name=talloc_strdup(mem_ctx, file->search.name))) goto error;
                        break;
        case RAW_SEARCH_DATA_STANDARD:
                        if (sws_dup(result, result->standard.name, file->standard.name)) goto error;
                        break;
        case RAW_SEARCH_DATA_EA_SIZE:
                        if (sws_dup(result, result->ea_size.name, file->ea_size.name)) goto error;
                        break;
        case RAW_SEARCH_DATA_EA_LIST:
                        if (sws_dup(result, result->ea_list.name, file->ea_list.name)) goto error;
                        break;
        case RAW_SEARCH_DATA_DIRECTORY_INFO:
                        if (sws_dup(result, result->directory_info.name, file->directory_info.name)) goto error;
                        break;
        case RAW_SEARCH_DATA_FULL_DIRECTORY_INFO:
                        if (sws_dup(result, result->full_directory_info.name, file->full_directory_info.name)) goto error;
                        break;
        case RAW_SEARCH_DATA_NAME_INFO:
                        if (sws_dup(result, result->name_info.name, file->name_info.name)) goto error;
                        break;
        case RAW_SEARCH_DATA_BOTH_DIRECTORY_INFO:
                        if (sws_dup(result, result->both_directory_info.name, file->both_directory_info.name)) goto error;
                        if (sws_dup(result, result->both_directory_info.short_name, file->both_directory_info.short_name)) goto error;
                        break;
        case RAW_SEARCH_DATA_ID_FULL_DIRECTORY_INFO:
                        if (sws_dup(result, result->id_full_directory_info.name, file->id_full_directory_info.name)) goto error;
                        break;
        case RAW_SEARCH_DATA_ID_BOTH_DIRECTORY_INFO:
                        if (sws_dup(result, result->id_both_directory_info.name, file->id_both_directory_info.name)) goto error;
                        if (sws_dup(result, result->id_both_directory_info.short_name, file->id_both_directory_info.short_name)) goto error;
                        break;
        case RAW_SEARCH_DATA_UNIX_INFO:
                        if (! (result->unix_info.name=talloc_strdup(mem_ctx, file->unix_info.name))) goto error;
                        break;
        case RAW_SEARCH_DATA_UNIX_INFO2:
                        if (sws_dup(result, result->unix_info2.name, file->unix_info2.name)) goto error;
                        break;
        default:
                        DEBUG(5,("%s: Error can't dup an unknown file data type: %x\n", __LOCATION__, data_level));
                        goto error;
        }
        return result;
error:
        talloc_free(result);
        return NULL;
}

/* callback function for search first/next */
static bool find_callback(void *private, const union smb_search_data *file)
{
        struct search_state *state = (struct search_state *)private;
        struct search_handle *search_handle = state->search_handle;
        bool status;

        /* if we have a cache, copy this data */
        if (search_handle->cache) {
                struct search_cache_item *item = talloc_zero(search_handle->cache, struct search_cache_item);
                DEBUG(5,("%s: Copy %p to cache %p\n", __LOCATION__, item, search_handle->cache));
                if (item) {
                        item->data_level=search_handle->data_level;
                        item->file = smb_search_data_dup(item, file, item->data_level);
                        if (! item->file) {
                                talloc_free(item);
                                item=NULL;
                        }
                }
                if (item) {
                        /* optimization to save enumerating the entire list each time, to find the end.
                           the cached last_item is very short lived, it doesn't matter if something has
                           been added since, as long as it hasn't been removed */
                        if (state->last_item) {
                                DLIST_ADD_END(state->last_item, item, struct search_cache_item*);
                        } else {
                                DLIST_ADD_END(search_handle->cache->items, item, struct search_cache_item*);
                        }
                        state->last_item=item;
                        state->all_count++;
                } else {
                        DEBUG(5,("%s: Could not add name to search cache %p, invalidating cache\n", __LOCATION__, search_handle->cache));
                        /* dear me, the whole cache will be invalid if we miss data */
                        search_handle->cache->status=SEARCH_CACHE_DEAD;
                        /* remove from the list of caches to use */
                        DLIST_REMOVE(search_handle->cache->proxy->search_caches, search_handle->cache);
                        /* Make it feel unwanted */
                        talloc_unlink(private, search_handle->cache);
                        talloc_unlink(search_handle, search_handle->cache);
                        //if (talloc_unlink(search_handle, search_handle->cache)==0) {
                                //talloc_free(search_handle->cache);
                        //}
                        /* stop us using it for this search too */
                        search_handle->cache=NULL;
                }
        }

        status=state->callback(state->private, file);
        if (status) {
                state->count++;
        }
        return status;
}

/*
   list files in a directory matching a wildcard pattern
*/
static NTSTATUS proxy_search_first(struct ntvfs_module_context *ntvfs,
                                  struct ntvfs_request *req, union smb_search_first *io,
                                  void *search_private,
                                  bool (*callback)(void *, const union smb_search_data *))
{
        struct proxy_private *private = ntvfs->private_data;
        struct search_state *state;
        struct search_cache *search_cache=NULL;
        struct search_cache_key search_cache_key={0};
        struct ntvfs_handle *h=NULL;
        struct search_handle *s;
        uint16_t max_count;
        NTSTATUS status;

        SETUP_PID;

        if (! private->enabled_proxy_search) {
                return smb_raw_search_first(private->tree, req, io, search_private, callback);
        }
        switch (io->generic.level) {
/*      case RAW_SEARCH_DATA_SEARCH:
                search_cache_key.search_attrib=io->search_first.in.search_attrib;
                search_cache_key.pattern=io->search_first.in.pattern;
                max_count = io->search_first.in.max_count;
                search_cache = find_search_cache(private->search_cache, &search_cache_key);
                break;*/
        case RAW_SEARCH_TRANS2:
                io->t2ffirst.in.max_count=MIN(io->t2ffirst.in.max_count,80);
                max_count = io->t2ffirst.in.max_count;

                search_cache_key.level=io->generic.level;
                search_cache_key.data_level=io->generic.data_level;
                search_cache_key.search_attrib=io->t2ffirst.in.search_attrib;
                search_cache_key.pattern=io->t2ffirst.in.pattern;
                search_cache_key.flags=io->t2ffirst.in.flags;
                search_cache_key.storage_type=io->t2ffirst.in.storage_type;
                /* try and find a search cache that is complete */
                search_cache = find_search_cache(private->search_caches, &search_cache_key);

                /* do handle mapping for TRANS2 */
                status = ntvfs_handle_new(ntvfs, req, &h);
                NT_STATUS_NOT_OK_RETURN(status);

                DEBUG(5,("%s: RAW_SEARCH_TRANS2 %s limit %d, cache=%p level=%x\n",__LOCATION__, search_cache_key.pattern, max_count, search_cache, search_cache_key.data_level));
                break;
        default: /* won't cache or proxy this */
                return smb_raw_search_first(private->tree, req, io, search_private, callback);
        }

        /* finish setting up mapped handle */
        if (h) {
                s = talloc_zero(h, struct search_handle);
                NT_STATUS_HAVE_NO_MEMORY(s);
                s->proxy=private;
                talloc_set_destructor(s, search_handle_destructor);
                s->h=h;
                s->level=io->generic.level;
                s->data_level=io->generic.data_level;
                status = ntvfs_handle_set_backend_data(s->h, private->ntvfs, s);
                NT_STATUS_NOT_OK_RETURN(status);
                DLIST_ADD(private->search_handles, s);
                DEBUG(5,("%s: map handle create %d\n",__LOCATION__, smbsrv_fnum(h)));
        }

        /* satisfy from cache */
        if (search_cache) {
                struct search_cache_item* item=search_cache->items;
                uint16_t count=0;

                /* stop cache going away while we are using it */
                s->cache = talloc_reference(s, search_cache);
                DEBUG(5,("%s: Serving from cache: %p\n",__LOCATION__, search_cache));
                /* Don't offer over the limit, but only count those that were accepted */
                DLIST_FIND(search_cache->items, item, !(count < max_count && callback(search_private, item->file) && ++count) );
                io->t2ffirst.out.count=count;
                s->resume_item=item;
                /* just because callback didn't accept any doesn't mean we are finished */
                if (item == NULL) {
                        /* currently only caching for t2ffirst */
                        io->t2ffirst.out.end_of_search = true;
                        DEBUG(5,("%s: Serving from cache complete at %d\n", __LOCATION__, count));
                } else {
                        /* count the rest */
                        io->t2ffirst.out.end_of_search = false;
                        DEBUG(5,("%s: Serving from cache incomplete at %d\n", __LOCATION__, count));
                        DLIST_FOR_EACH(item, item, count++);
                        DEBUG(5,("%s: Serving from cache max_count %d\n", __LOCATION__, count));
                }

                if ((io->t2ffirst.out.end_of_search && io->t2ffirst.in.flags & FLAG_TRANS2_FIND_CLOSE_IF_END) ||
                    io->t2ffirst.in.flags & FLAG_TRANS2_FIND_CLOSE)
                {
                        /* destroy handle */
                        ntvfs_handle_remove_backend_data(h, ntvfs);
                        io->t2ffirst.out.handle=0;
                } else {
                        /* now map handle */
                        io->t2ffirst.out.handle=smbsrv_fnum(h);
                }
                return NT_STATUS_OK;
        }

        state = talloc_zero(req, struct search_state);
        NT_STATUS_HAVE_NO_MEMORY(state);

        /* if there isn't a matching cache already being generated by another search,
           start one, unless FLAG_TRANS2_FIND_BACKUP_INTENT which is always live */
        if (!(io->t2ffirst.in.flags & FLAG_TRANS2_FIND_BACKUP_INTENT) &&
            find_partial_search_cache(private->search_caches, &search_cache_key) == NULL) {
                /* need to opendir the folder being searched so we can get a notification */
                struct search_cache *search_cache=NULL;

                search_cache=new_search_cache(private, &search_cache_key);
                /* Stop cache going away while we are using it */
                if (search_cache) {
                        s->cache=talloc_reference(s, search_cache);
                }
        }

        /* stop the handle going away while we are using it */
        state->search_handle=talloc_reference(state, s);
        state->private=search_private;
        state->callback=callback;

        status=smb_raw_search_first(private->tree, req, io, state, find_callback);
//      if (! NT_STATUS_IS_OK(status)) {
//              return (status);
//      }
        if (! NT_STATUS_IS_OK(status)) {
                if (s->cache) {
                        DLIST_REMOVE(private->search_caches, s->cache);
                        talloc_unlink(private, s->cache);
                        talloc_unlink(s, s->cache);
                        //if (talloc_unlink(s, s->cache)==0) {
                                //talloc_free(s->cache);
                        //}
                        s->cache=NULL;
                }
                s->h=NULL;
                ntvfs_handle_remove_backend_data(h, ntvfs);
                return (status);
        }
//      DEBUG(1,("%s: %p; %s\n",__LOCATION__,io,get_friendly_nt_error_msg (status)));
        DEBUG(5,("%s: max %d, got %d, copied %d; %s\n",__LOCATION__,io->t2ffirst.out.count,state->all_count, state->count,get_friendly_nt_error_msg (status)));

#warning check NT_STATUS_IS_OK ?
        if (io->t2ffirst.out.end_of_search) {
                /* cache might have gone away if problem filling */
                if (s->cache) {
                    DEBUG(5,("B\n"));
                        s->cache->status = SEARCH_CACHE_COMPLETE;
                        DEBUG(5,("%s: Cache %p filled in first go!\n",__LOCATION__, s->cache));
                }
        }
        if ((io->t2ffirst.out.end_of_search && io->t2ffirst.in.flags & FLAG_TRANS2_FIND_CLOSE_IF_END) ||
            io->t2ffirst.in.flags & FLAG_TRANS2_FIND_CLOSE) {
                DEBUG(5,("%s: Closing search\n",__LOCATION__));
                /* destroy partial cache */
                if (s->cache && (io->t2ffirst.in.flags & FLAG_TRANS2_FIND_CLOSE) &&
                    ! io->t2ffirst.out.end_of_search) {
                        DEBUG(5,("%s: Destroying cache %p\n",__LOCATION__, s->cache));
                        /* cache is no good now! */
                        DLIST_REMOVE(private->search_caches, s->cache);
                        talloc_unlink(private, s->cache);
                        talloc_unlink(s, s->cache);
                        //if (talloc_unlink(s, s->cache)==0) {
                                //talloc_free(s->cache);
                        //}
                        s->cache=NULL;
                }
                if (s->cache) {
                        s->cache->status=SEARCH_CACHE_COMPLETE;
                }
                /* Need to deal with the case when the client would not take them all but we still cache them
                if (state->count < io->t2ffirst.out.count && io->t2ffirst.out.end_of_search) {
                        io->t2ffirst.out.end_of_search = false;
                        //s->resume_item = state->last_item;
                }*/
                /* destroy handle */
                DEBUG(5,("%s: Removing handle %p\n",__LOCATION__,h));
                ntvfs_handle_remove_backend_data(h, ntvfs);
                io->t2ffirst.out.handle=0;
        } else {
                s->handle = io->t2ffirst.out.handle;
                io->t2ffirst.out.handle=smbsrv_fnum(h);
        }
        io->t2ffirst.out.count=state->count;
        return status;
}

#define DLIST_FIND_NEXT(start, item, test) do {\
DLIST_FIND(start, item, test); \
if (item) (item)=(item)->next; \
} while(0)
#define DLIST_TALLOC_FREE(list) do {\
        while(list) { \
                void *tmp=(list); \
                (list)=(list)->next; \
                talloc_free(tmp); \
        } \
} while(0)

/* continue a search */
static NTSTATUS proxy_search_next(struct ntvfs_module_context *ntvfs,
                                 struct ntvfs_request *req, union smb_search_next *io,
                                 void *search_private,
                                 bool (*callback)(void *, const union smb_search_data *))
{
        struct proxy_private *private = ntvfs->private_data;
        struct search_state *state;
        struct ntvfs_handle *h=NULL;
        struct search_handle *s;
        const struct search_cache *search_cache=NULL;
        struct search_cache_item *start_at=NULL;
        uint16_t max_count;
        NTSTATUS status;

        SETUP_PID;

        if (! private->enabled_proxy_search) {
                return smb_raw_search_next(private->tree, req, io, search_private, callback);
        }
        switch (io->generic.level) {
        case RAW_SEARCH_TRANS2:
                io->t2fnext.in.max_count=MIN(io->t2fnext.in.max_count,80);
                max_count = io->t2fnext.in.max_count;

                h = talloc_get_type(ntvfs_find_handle(ntvfs, req, io->t2fnext.in.handle), struct ntvfs_handle);
                if (! h) return NT_STATUS_INVALID_HANDLE;
                /* convert handle into search_cache */
                s=talloc_get_type(ntvfs_handle_get_backend_data(h, ntvfs), struct search_handle);
                if (! s) return NT_STATUS_INVALID_HANDLE;
                search_cache=s->cache;
                DEBUG(5,("%s: RAW_SEARCH_TRANS2 find_next h=%d [real %d] count %d, cache=%p\n",__LOCATION__, io->t2fnext.in.handle, s->handle, max_count, search_cache));
                io->t2fnext.in.handle=s->handle;
                if (! search_cache) {
                        break;
                }

                /* warning if: uint16_t flags or const char *last_name have changed, abort cache */
                /* skip up to resume key */
                /* TODO: resume key may be PRIOR to where we left off... in which case
                   we need to avoid duplicating values */
                if (search_cache /*&& search_cache->status == SEARCH_CACHE_COMPLETE*/) {
                        DEBUG(5,("%s: seek resume position\n",__LOCATION__));
                        /* work out where in the cache to continue from */
                        switch (io->generic.data_level) {
                        case RAW_SEARCH_DATA_STANDARD:
                        case RAW_SEARCH_DATA_EA_SIZE:
                        case RAW_SEARCH_DATA_EA_LIST:
                                /* have a resume key? */
                                DEBUG(5,("%s: type %x seek on %x\n",__LOCATION__, io->generic.data_level, io->t2fnext.in.resume_key));
                                DLIST_FIND_NEXT(search_cache->items, start_at, io->t2fnext.in.resume_key == start_at->file->standard.resume_key);
                                break;
                        case RAW_SEARCH_DATA_DIRECTORY_INFO: /* TODO: maybe these should be strcasecmp for some filesystems */
                                DEBUG(5,("%s: type %x seek on %s\n",__LOCATION__, io->generic.data_level, io->t2fnext.in.last_name));
                                DLIST_FIND_NEXT(search_cache->items, start_at, fstrcmp(io->t2fnext.in.last_name, start_at->file->directory_info.name.s)==0);
                                break;
                        case RAW_SEARCH_DATA_FULL_DIRECTORY_INFO:
                                DEBUG(5,("%s: type %x seek on %s\n",__LOCATION__, io->generic.data_level, io->t2fnext.in.last_name));
                                DLIST_FIND_NEXT(search_cache->items, start_at, fstrcmp(io->t2fnext.in.last_name, start_at->file->full_directory_info.name.s)==0);
                                break;
                        case RAW_SEARCH_DATA_NAME_INFO:
                                DEBUG(5,("%s: type %x seek on %s\n",__LOCATION__, io->generic.data_level, io->t2fnext.in.last_name));
                                DLIST_FIND_NEXT(search_cache->items, start_at, fstrcmp(io->t2fnext.in.last_name, start_at->file->name_info.name.s)==0);
                                break;
                        case RAW_SEARCH_DATA_BOTH_DIRECTORY_INFO:
                                DEBUG(5,("%s: type %x seek on %s\n",__LOCATION__, io->generic.data_level, io->t2fnext.in.last_name));
                                DLIST_FIND_NEXT(search_cache->items, start_at, fstrcmp(io->t2fnext.in.last_name, start_at->file->both_directory_info.name.s)==0);
                                break;
                        case RAW_SEARCH_DATA_ID_FULL_DIRECTORY_INFO:
                                DEBUG(5,("%s: type %x seek on %s\n",__LOCATION__, io->generic.data_level, io->t2fnext.in.last_name));
                                DLIST_FIND_NEXT(search_cache->items, start_at, fstrcmp(io->t2fnext.in.last_name, start_at->file->id_full_directory_info.name.s)==0);
                                break;
                        case RAW_SEARCH_DATA_ID_BOTH_DIRECTORY_INFO:
                                DEBUG(5,("%s: type %x seek on %s\n",__LOCATION__, io->generic.data_level, io->t2fnext.in.last_name));
                                DLIST_FIND_NEXT(search_cache->items, start_at, fstrcmp(io->t2fnext.in.last_name, start_at->file->id_both_directory_info.name.s)==0);
                                break;
                        case RAW_SEARCH_DATA_UNIX_INFO:
                                DEBUG(5,("%s: type %x seek on %s\n",__LOCATION__, io->generic.data_level, io->t2fnext.in.last_name));
                                DLIST_FIND_NEXT(search_cache->items, start_at, fstrcmp(io->t2fnext.in.last_name, start_at->file->unix_info.name)==0);
                                break;
                        case RAW_SEARCH_DATA_UNIX_INFO2:
                                DEBUG(5,("%s: type %x seek on %s\n",__LOCATION__, io->generic.data_level, io->t2fnext.in.last_name));
                                DLIST_FIND_NEXT(search_cache->items, start_at, fstrcmp(io->t2fnext.in.last_name, start_at->file->unix_info2.name.s)==0);
                                break;
                        default:
                                if (io->t2fnext.in.flags & FLAG_TRANS2_FIND_CONTINUE) {
                                        start_at = s->resume_item;
                                } else {
                                        DEBUG(5,("%s: HELP! How can we resume?\n",__LOCATION__));
                                        start_at = s->resume_item;
                                }
                        }
                        DEBUG(5,("%s: Start at %p\n",__LOCATION__,start_at));
                }
                break;
        }

        if (! search_cache) {
                DEBUG(5,("%s: No cache, pass-through\n",__LOCATION__));
                return smb_raw_search_next(private->tree, req, io, search_private, callback);
        }
//#define talloc_reference(ctx, ptr) (_TALLOC_TYPEOF(ptr))_talloc_reference((ctx),(ptr))
//surely should be
//#define talloc_reference(ctx, ptr) _talloc_reference((ctx),(ptr))?(ptr):(NULL) to preserve the type of ptr

        /* satisfy from cache */
        if (search_cache->status == SEARCH_CACHE_COMPLETE) {
                struct search_cache_item* item;
                uint16_t count=0;
                DEBUG(5,("%s: Serving from cache: %p\n",__LOCATION__, search_cache));

                if (! start_at) {
                        start_at = search_cache->items;
                }

                DLIST_FIND(start_at, item, !(count < max_count && callback(search_private, item->file) && ++count) );
                io->t2fnext.out.count=count;
                s->resume_item=item;
                if (item == NULL) {
                        DEBUG(5,("%s: Serving from cache complete at %d\n", __LOCATION__, count));
                        io->t2fnext.out.end_of_search = true;
                } else {
                        DEBUG(5,("%s: Serving from cache incomplete at %d\n", __LOCATION__, count));
                        io->t2fnext.out.end_of_search = false;
                        /* count the rest */
                        DLIST_FOR_EACH(item, item, count++);
                        DEBUG(5,("%s: Serving from cache max_count %d\n", __LOCATION__, count));
                }
                /* is it the end? */
                if ((io->t2fnext.out.end_of_search && io->t2fnext.in.flags & FLAG_TRANS2_FIND_CLOSE_IF_END) ||
                        io->t2fnext.in.flags & FLAG_TRANS2_FIND_CLOSE)
                {
                        /* destroy handle */
                        DEBUG(5,("%s: Removing handle %p\n",__LOCATION__,h));
                        ntvfs_handle_remove_backend_data(h, ntvfs);
                }

                return NT_STATUS_OK;
        }

        /* pass-through and fill-cache */
        if (start_at) {
                /* risk of duplicate data */
                DEBUG(5,("\n\n\nCache-populating search has resumed but NOT where we left off!\n\n\n-d"));
                /* free everything from start_at onwards through start_at-> next*/
                /* cut from the list */
                start_at->prev->next=NULL;
                start_at->prev=NULL;
                /* now how to free a list? */
                DLIST_TALLOC_FREE(start_at);
        }
        state = talloc_zero(req, struct search_state);
        NT_STATUS_HAVE_NO_MEMORY(state);

        state->search_handle=talloc_reference(state, s);
        state->private=search_private;
        state->callback=callback;

        status = smb_raw_search_next(private->tree, req, io, state, find_callback);
        if (! NT_STATUS_IS_OK(status)) {
                if (s->cache) {
                        DLIST_REMOVE(private->search_caches, s->cache);
                        talloc_unlink(private, s->cache);
                        talloc_unlink(s, s->cache);
                        //if (talloc_unlink(s, s->cache)==0) {
                                //talloc_free(s->cache);
                        //}
                        s->cache=NULL;
                }
                s->h=NULL;
                ntvfs_handle_remove_backend_data(h, ntvfs);
                return (status);
        }

        DEBUG(5,("%s: max %d, got %d, copied %d; %s\n",__LOCATION__,io->t2fnext.out.count,state->all_count, state->count,get_friendly_nt_error_msg (status)));

        /* if closing, then close */
        if ((io->t2fnext.out.end_of_search && io->t2fnext.in.flags & FLAG_TRANS2_FIND_CLOSE_IF_END) ||
                io->t2fnext.in.flags & FLAG_TRANS2_FIND_CLOSE)
        {
                if (s->cache && (io->t2fnext.in.flags & FLAG_TRANS2_FIND_CLOSE) &&
                    ! io->t2fnext.out.end_of_search) {
                        /* partial cache is useless */
                        DLIST_REMOVE(private->search_caches, s->cache);
                        talloc_unlink(private, s->cache);
                        talloc_unlink(s, s->cache);
                        //if (talloc_unlink(s, s->cache)==0) {
                                //talloc_free(s->cache);
                        //}
                        s->cache=NULL;
                }
                if (s->cache) {
                        s->cache->status=SEARCH_CACHE_COMPLETE;
                        /* Need to deal with the case when the client would not take them all but we still cache them
                        if (state->count < io->t2fnext.out.count && io->t2fnext.out.end_of_search) {
                                io->t2fnext.out.end_of_search = false;
                        }*/
                }
                /* destroy handle */
                DEBUG(5,("%s: Removing handle %p\n",__LOCATION__,h));
                ntvfs_handle_remove_backend_data(h, ntvfs);
        }
        io->t2fnext.out.count=state->count;

        return status;
}

/* close a search */
static NTSTATUS proxy_search_close(struct ntvfs_module_context *ntvfs,
                                  struct ntvfs_request *req, union smb_search_close *io)
{
        struct proxy_private *private = ntvfs->private_data;
        struct ntvfs_handle *h=NULL;
        struct search_handle *s;
        NTSTATUS status;

        SETUP_PID;

        if (! private->enabled_proxy_search) {
                return smb_raw_search_close(private->tree, io);
        }
        switch (io->generic.level) {
        case RAW_SEARCH_TRANS2:
                h = talloc_get_type(ntvfs_find_handle(ntvfs, req, io->findclose.in.handle), struct ntvfs_handle);
                if (! h) return NT_STATUS_INVALID_HANDLE;
                /* convert handle into search_cache */
                s=talloc_get_type(ntvfs_handle_get_backend_data(h, ntvfs), struct search_handle);
                if (! s) return NT_STATUS_INVALID_HANDLE;
                io->findclose.in.handle=s->handle;
        default:
                return smb_raw_search_close(private->tree, io);
        }

        if (! s->cache) {
                status = smb_raw_search_close(private->tree, io);
        } else {
                if (s->cache->status != SEARCH_CACHE_COMPLETE) {
                        /* cache is useless */
                        DLIST_REMOVE(private->search_caches, s->cache);
                        talloc_unlink(private, s->cache);
                        talloc_unlink(s, s->cache);
                        //if (talloc_unlink(s, s->cache)==0) {
                                //talloc_free(s->cache);
                        //}
                }
                status = NT_STATUS_OK;
        }

        s->h=NULL;
        ntvfs_handle_remove_backend_data(h, ntvfs);
        /* s MAY also be gone at this point, if h was free'd, unless there were
           pending responses, in which case they see s->h is NULL as a sign to stop */
        return status;
}

/*
  a handler for async trans2 replies
 */
static void async_trans2(struct smbcli_request *c_req)
{
        struct async_info *async = c_req->async.private;
        struct ntvfs_request *req = async->req;
        req->async_states->status = smb_raw_trans2_recv(c_req, req, async->parms);
        talloc_free(async);
        req->async_states->send_fn(req);
}

/* raw trans2 */
static NTSTATUS proxy_trans2(struct ntvfs_module_context *ntvfs,
                            struct ntvfs_request *req,
                            struct smb_trans2 *trans2)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;

        if (private->map_trans2) {
                return NT_STATUS_NOT_IMPLEMENTED;
        }

        SETUP_PID;
#warning we should be mapping file handles here

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return smb_raw_trans2(private->tree, req, trans2);
        }

        c_req = smb_raw_trans2_send(private->tree, trans2);

        ASYNC_RECV_TAIL(trans2, async_trans2);
}


/* SMBtrans - not used on file shares */
static NTSTATUS proxy_trans(struct ntvfs_module_context *ntvfs,
                           struct ntvfs_request *req,
                           struct smb_trans2 *trans2)
{
        return NT_STATUS_ACCESS_DENIED;
}

/*
  a handler for async change notify replies
 */
static void async_changenotify(struct smbcli_request *c_req)
{
        struct async_info *async = c_req->async.private;
        struct ntvfs_request *req = async->req;
        req->async_states->status = smb_raw_changenotify_recv(c_req, req, async->parms);
        talloc_free(async);
        req->async_states->send_fn(req);
}

/* change notify request - always async */
static NTSTATUS proxy_notify(struct ntvfs_module_context *ntvfs,
                            struct ntvfs_request *req,
                            union smb_notify *io)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;
        int saved_timeout = private->transport->options.request_timeout;
        struct proxy_file *f;

        if (io->nttrans.level != RAW_NOTIFY_NTTRANS) {
                return NT_STATUS_NOT_IMPLEMENTED;
        }

        SETUP_PID;

        f = ntvfs_handle_get_backend_data(io->nttrans.in.file.ntvfs, ntvfs);
        if (!f) return NT_STATUS_INVALID_HANDLE;
        io->nttrans.in.file.fnum = f->fnum;

        /* this request doesn't make sense unless its async */
        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return NT_STATUS_INVALID_PARAMETER;
        }

        /* we must not timeout on notify requests - they wait
           forever */
        private->transport->options.request_timeout = 0;

        c_req = smb_raw_changenotify_send(private->tree, io);

        private->transport->options.request_timeout = saved_timeout;

        ASYNC_RECV_TAIL(io, async_changenotify);
}

/*
 * A hander for converting from rpc struct replies to ntioctl
 */
static NTSTATUS proxy_rpclite_map_async_send(
                                        struct ntvfs_module_context *ntvfs,
                                   struct ntvfs_request *req,
                                   void *io1, void *io2, NTSTATUS status)
{
        union smb_ioctl* io=talloc_get_type_abort(io1, union smb_ioctl);
        struct async_rpclite_send *rpclite_send=talloc_get_type_abort(io2, struct async_rpclite_send);
        void* r=rpclite_send->struct_ptr;
        struct ndr_push* push;
        const struct ndr_interface_call* call=rpclite_send->call;
        enum ndr_err_code ndr_err;
        DATA_BLOB ndr;

        talloc_free(rpclite_send);

        DEBUG(5,("%s: converting r=%p back to ntiocl\n",__FUNCTION__, r));
        push = ndr_push_init_ctx(req, lp_iconv_convenience(ntvfs->ctx->lp_ctx));
        NT_STATUS_HAVE_NO_MEMORY(push);

        if (0) {
                push->flags |= LIBNDR_FLAG_BIGENDIAN;
        }

        ndr_err = call->ndr_push(push, NDR_OUT, r);
        status=ndr_map_error2ntstatus(ndr_err);

        if (! NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
                DEBUG(2,("Unable to ndr_push structure in dcerpc_ndr_request_send - %s\n",
                        nt_errstr(status)));
                return status;
        }

        ndr=ndr_push_blob(push);
        //if (ndr.length > io->ntioctl.in.max_data) {
        DEBUG(3,("%s NDR size %d, max_size %d %p\n",__FUNCTION__, ndr.length,
                         io->ntioctl.in.max_data, ndr.data));
        io->ntioctl.out.blob=ndr;
        return status;
}

/*
 * A handler for sending async rpclite Read replies that were mapped to union smb_read
 */
static NTSTATUS rpclite_proxy_Read_map_async_send(
                                        struct ntvfs_module_context *ntvfs,
                                        struct ntvfs_request *req,
                                        void *io1, void *io2, NTSTATUS status)
{
        struct proxy_Read* r=talloc_get_type_abort(io1, struct proxy_Read);
        union smb_read* io=talloc_get_type_abort(io2, union smb_read);

        /* status here is a result of proxy_read, it doesn't reflect the status
           of the rpc transport or relates calls, just the read operation */
        DEBUG(5,("%s with jolly status %s\n",__FUNCTION__, get_friendly_nt_error_msg(status)));
        r->out.result=status;

        if (! NT_STATUS_IS_OK(status)) {
                /* We can't use result as a discriminator in IDL, so nread and flags always exist */
                r->out.nread=0;
                r->out.flags=0;
        } else {
                ssize_t size=io->readx.out.nread;
                r->out.flags=0;
                r->out.nread=io->readx.out.nread;

                if (r->in.flags & (PROXY_USE_CACHE | PROXY_VALIDATE) && io->readx.out.nread>0) {
                        declare_checksum(digest);
                        checksum_block(digest, io->readx.out.data, io->readx.out.nread);

                        DEBUG(5,("New digest for size: %lld\n", (long long) io->readx.out.nread));
                        dump_data (5, digest, sizeof(digest));
                        DEBUG(5,("Cached digest\n"));
                        dump_data (5, r->in.digest.digest, sizeof(digest));

                        if (memcmp(digest, r->in.digest.digest, sizeof(digest))==0) {
                                r->out.flags=PROXY_USE_CACHE;
                                DEBUG(5,("%s: Use cached data len=%lld\n",__FUNCTION__,
                                                 (long long)r->out.nread));
                                if (r->in.flags & PROXY_VALIDATE) {
                                        r->out.flags |= PROXY_VALIDATE;
                                        DEBUG(5,("%s: Use VALIDATED  len=%lld, %lld\n",__FUNCTION__,
                                                 (long long)r->out.nread, (long long) io->readx.out.nread));
                                }
                                goto done;
                        }
                        DEBUG(5,("Cache does not match\n"));
                }

                if (r->in.flags & PROXY_VALIDATE) {
                        /* validate failed, shrink read to mincnt - so we don't fill link */
                        r->out.nread=MIN(r->out.nread, r->in.mincnt);
                        size=r->out.nread;
                        DEBUG(5,("VALIDATE failed, shrink read of %d from %d to %d\n",
                                         r->in.maxcnt,r->out.nread,MIN(r->out.nread, r->in.mincnt)));
                }

                if (r->in.flags & PROXY_USE_ZLIB) {
                        if (compress_block(io->readx.out.data, &size) ) {
                                r->out.flags|=PROXY_USE_ZLIB;
                                r->out.response.compress.count=size;
                                r->out.response.compress.data=io->readx.out.data;
                                DEBUG(3,("%s: Compressed from %d to %d = %d%%\n",
                                        __FUNCTION__,r->out.nread,size,size*100/r->out.nread));
                                goto done;
                        }
                }

                DEBUG(5,("%s: Compression not worthwhile\n", __FUNCTION__));
                r->out.response.generic.count=io->readx.out.nread;
                r->out.response.generic.data=io->readx.out.data;
        }

done:

        /* Or should we return NT_STATUS_OK ?*/
        DEBUG(5,("Finish %s status %s\n",__FUNCTION__,get_friendly_nt_error_msg(status)));

        /* the rpc transport succeeded even if the operation did not */
        return NT_STATUS_OK;
}

/*
 * RPC implementation of Read
 */
static NTSTATUS rpclite_proxy_Read(struct ntvfs_module_context *ntvfs,
                           struct ntvfs_request *req, struct proxy_Read *r)
{
        struct proxy_private *private = ntvfs->private_data;
        union smb_read* io=talloc(req, union smb_read);
        NTSTATUS status;
        struct proxy_file *f;
        struct ntvfs_handle *h;

        NT_STATUS_HAVE_NO_MEMORY(io);

/* if next hop is a proxy just repeat this call also handle VALIDATE check
   that means have own callback handlers too... */
        SETUP_PID;

        RPCLITE_SETUP_FILE_HERE(f, h);

        DEBUG(5,("Opnum: proxy_Read min=%d max=%d offset=%lld, fnum=%d\n",
                         r->in.mincnt, r->in.maxcnt, r->in.offset, r->in.fnum));
        DEBUG(5,("Anticipated digest\n"));
        dump_data (5, r->in.digest.digest, sizeof(r->in.digest.digest));

/*      If the remove end is a proxy, jusr fixup file handle and passthrough,
        but update cache on the way back
        if (PROXY_REMOTE_SERVER(private) && (r->in.flags & PROXY_VALIDATE)) {
        }
*/
        /* prepare for response */
        r->out.response.generic.data=talloc_array(io, uint8_t, r->in.maxcnt);
        NT_STATUS_HAVE_NO_MEMORY(r->out.response.generic.data);

        if (! PROXY_REMOTE_SERVER(private) && (r->in.flags & PROXY_VALIDATE)) {
                return proxy_validate(ntvfs, req, r, f);
        }

        /* pack up an smb_read request and dispatch here */
        io->readx.level=RAW_READ_READX;
        io->readx.in.file.ntvfs=h;
        io->readx.in.mincnt=r->in.mincnt;
        io->readx.in.maxcnt=r->in.maxcnt;
        io->readx.in.offset=r->in.offset;
        io->readx.in.remaining=r->in.remaining;
        /* and something to hold the answer */
        io->readx.out.data=r->out.response.generic.data;

        /* so we get to pack the io->*.out response */
        status = ntvfs_map_async_setup(ntvfs, req, r, io, rpclite_proxy_Read_map_async_send);
        NT_STATUS_NOT_OK_RETURN(status);

        /* so the read will get processed normally */
        return proxy_read(ntvfs, req, io);
}

/*
 * A handler for sending async rpclite Write replies
 */
static NTSTATUS rpclite_proxy_Write_map_async_send(
                                                                                struct ntvfs_module_context *ntvfs,
                                                                                struct ntvfs_request *req,
                                                                                void *io1, void *io2, NTSTATUS status)
{
        struct proxy_Write* r=talloc_get_type_abort(io1, struct proxy_Write);
        union smb_write* io=talloc_get_type_abort(io2, union smb_write);

        DEBUG(5,("%s with jolly status %s\n",__FUNCTION__, get_friendly_nt_error_msg(status)));
        r->out.result=status;

                r->out.nwritten=io->writex.out.nwritten;
                r->out.remaining=io->writex.out.remaining;

        /* the rpc transport succeeded even if the operation did not */
        return NT_STATUS_OK;
}

/*
 * RPC implementation of write
 */
static NTSTATUS rpclite_proxy_Write(struct ntvfs_module_context *ntvfs,
                                                        struct ntvfs_request *req, struct proxy_Write *r)
{
        struct proxy_private *private = ntvfs->private_data;
        union smb_write* io=talloc(req, union smb_write);
        NTSTATUS status;
        struct proxy_file* f;
        struct ntvfs_handle *h;

        SETUP_PID;

        RPCLITE_SETUP_FILE_HERE(f,h);

        DEBUG(5,("Opnum: proxy_Write count=%d offset=%lld, fnum=%d\n",
                                                r->in.count, r->in.offset, r->in.fnum));

        /* pack up an smb_write request and dispatch here */
        io->writex.level=RAW_WRITE_WRITEX;
        io->writex.in.file.ntvfs=h;
        io->writex.in.offset=r->in.offset;
        io->writex.in.wmode=r->in.mode;
        io->writex.in.count=r->in.count;

        /* and the data */
        if (PROXY_USE_ZLIB & r->in.flags) {
                ssize_t count=r->in.data.generic.count;
                io->writex.in.data=uncompress_block_talloc(io, r->in.data.compress.data,
                                                                                                        &count, r->in.count);
                if (count != r->in.count || !io->writex.in.data) {
                        /* Didn't uncompress properly, but the RPC layer worked */
                        r->out.result=NT_STATUS_BAD_COMPRESSION_BUFFER;
                        return NT_STATUS_OK;
                }
        } else {
                io->writex.in.data=r->in.data.generic.data;
        }

        /* so we get to pack the io->*.out response */
        status=ntvfs_map_async_setup(ntvfs, req, r, io, rpclite_proxy_Write_map_async_send);
        NT_STATUS_NOT_OK_RETURN(status);

        /* so the read will get processed normally */
        return proxy_write(ntvfs, req, io);
}

/*
 * RPC amalgamation of getinfo requests
 */
struct proxy_getinfo_fragments;
struct proxy_getinfo_fragmentses;

/* holds one smbcli_request to satisfy part of one proxy_GetInfo request */
struct proxy_getinfo_fragment {
        struct proxy_getinfo_fragment *prev, *next;
        struct proxy_getinfo_fragments *fragments;
        union smb_fileinfo *smb_fileinfo;
        struct smbcli_request *c_req;
        NTSTATUS status;
};

/* holds reference to many fragment smbcli_request that together make up one proxy_GetInfo request */
struct proxy_getinfo_fragments {
        struct proxy_getinfo_fragments *prev, *next;
        struct proxy_getinfo_fragmentses *fragmentses;
        struct proxy_getinfo_fragment *fragments;
        uint32_t index;
};

struct proxy_getinfo_fragmentses {
        struct proxy_getinfo_fragments *fragments;
        struct proxy_GetInfo *r;
        struct ntvfs_request *req;
        bool async;
};

/*
  a handler for async write replies
 */
NTSTATUS async_proxy_getinfo(struct async_info *async, void* io1, void* io2, NTSTATUS status)
{
        struct smbcli_request *c_req = async->c_req;
        struct ntvfs_request *req = async->req;
        struct proxy_file *f=async->f;
        struct proxy_getinfo_fragment *fragment=talloc_get_type_abort(io2, struct proxy_getinfo_fragment);
        struct proxy_getinfo_fragments* fragments=fragment->fragments;
        struct proxy_getinfo_fragmentses* fragmentses=fragments->fragmentses;
        struct proxy_GetInfo *r=talloc_get_type_abort(fragmentses->r, struct proxy_GetInfo);
        int c=fragments->index;
        struct info_data* d=&(r->out.info_data[c]);
        union smb_fileinfo *io=talloc_get_type_abort(io1, union smb_fileinfo);

        SMB_ASSERT(c_req == NULL || c_req == fragment->c_req);

        if (c_req) {
                switch (r->in.info_tags[0].tag_type) {
                case TAG_TYPE_FILE_INFO:
                        status=smb_raw_fileinfo_recv(c_req, r, io);
                        break;
                case TAG_TYPE_PATH_INFO:
                        status=smb_raw_pathinfo_recv(c_req, r, io);
                        break;
                default:
                        status=NT_STATUS_INVALID_PARAMETER;
                }
                c_req=NULL;
        }

        /* stop callback occuring more than once sync'ly */
        fragment->c_req=NULL;

        DEBUG(5,("%s: async callback level %x %s\n",__FUNCTION__,io->generic.level, get_friendly_nt_error_msg (status)));
        switch (io->generic.level) {
        case RAW_FILEINFO_ALL_INFO:
        case RAW_FILEINFO_ALL_INFORMATION:
                DEBUG(5,("%s: async callback level RAW_FILEINFO_ALL_INFO\n",__FUNCTION__));
                d->status_RAW_FILEINFO_ALL_INFO=status;

                /* don't blindly overwrite BASIC_INFORMATION as we may already have it */
                if (1 || NT_STATUS_IS_OK(status)) {
                        d->status_RAW_FILEINFO_BASIC_INFORMATION=status;
                        d->create_time=io->all_info.out.create_time;
                        d->access_time=io->all_info.out.access_time;
                        d->write_time=io->all_info.out.write_time;
                        d->change_time=io->all_info.out.change_time;
                        d->attrib=io->all_info.out.attrib;
                }
                d->alloc_size=io->all_info.out.alloc_size;
                d->size=io->all_info.out.size;
        dump_data(5, io, sizeof(*io));
                d->nlink=io->all_info.out.nlink;
                d->delete_pending=io->all_info.out.delete_pending;
                d->directory=io->all_info.out.directory;
                d->ea_size=io->all_info.out.ea_size;
                /* io is sticking around for as long as d is */
                d->fname.s=io->all_info.out.fname.s;
                d->fname.count=io->all_info.out.fname.private_length;
                break;
        case RAW_FILEINFO_BASIC_INFO:
        case RAW_FILEINFO_BASIC_INFORMATION:
                DEBUG(5,("%s: async callback level RAW_FILEINFO_BASIC_INFORMATION\n",__FUNCTION__));
                d->status_RAW_FILEINFO_BASIC_INFORMATION=status;
                d->create_time=io->basic_info.out.create_time;
                d->access_time=io->basic_info.out.access_time;
                d->write_time=io->basic_info.out.write_time;
                d->change_time=io->basic_info.out.change_time;
                d->attrib=io->basic_info.out.attrib;
                break;
        case RAW_FILEINFO_COMPRESSION_INFO:
                DEBUG(5,("%s: async callback level RAW_FILEINFO_COMPRESSION_INFO\n",__FUNCTION__));
                d->status_RAW_FILEINFO_COMPRESSION_INFO = status;
                d->compressed_size=io->compression_info.out.compressed_size;
                d->format=io->compression_info.out.format;
                d->unit_shift=io->compression_info.out.unit_shift;
                d->chunk_shift=io->compression_info.out.chunk_shift;
                d->cluster_shift=io->compression_info.out.cluster_shift;
                break;
        case RAW_FILEINFO_INTERNAL_INFORMATION:
                DEBUG(5,("%s: async callback level RAW_FILEINFO_INTERNAL_INFORMATION\n",__FUNCTION__));
                d->status_RAW_FILEINFO_INTERNAL_INFORMATION=status;
                d->file_id=io->internal_information.out.file_id;
                break;
        case RAW_FILEINFO_ACCESS_INFORMATION:
                DEBUG(5,("%s: async callback level RAW_FILEINFO_ACCESS_INFORMATION\n",__FUNCTION__));
                d->status_RAW_FILEINFO_ACCESS_INFORMATION=status;
                d->access_flags=io->access_information.out.access_flags;
                break;
        case RAW_FILEINFO_POSITION_INFORMATION:
                DEBUG(5,("%s: async callback level RAW_FILEINFO_POSITION_INFORMATION\n",__FUNCTION__));
                d->status_RAW_FILEINFO_POSITION_INFORMATION = status;
                d->position=io->position_information.out.position;
                break;
        case RAW_FILEINFO_MODE_INFORMATION:
                DEBUG(5,("%s: async callback level RAW_FILEINFO_MODE_INFORMATION\n",__FUNCTION__));
                d->status_RAW_FILEINFO_MODE_INFORMATION =status;
                d->mode=io->mode_information.out.mode;
                break;
        case RAW_FILEINFO_ALIGNMENT_INFORMATION:
                DEBUG(5,("%s: async callback level RAW_FILEINFO_ALIGNMENT_INFORMATION\n",__FUNCTION__));
                d->status_RAW_FILEINFO_ALIGNMENT_INFORMATION=status;
                d->alignment_requirement=io->alignment_information.out.alignment_requirement;
                break;
        case RAW_FILEINFO_ATTRIBUTE_TAG_INFORMATION:
                DEBUG(5,("%s: async callback level RAW_FILEINFO_ATTRIBUTE_TAG_INFORMATION\n",__FUNCTION__));
                d->status_RAW_FILEINFO_ATTRIBUTE_TAG_INFORMATION=status;
                d->reparse_tag=io->attribute_tag_information.out.reparse_tag;
                d->reparse_attrib=io->attribute_tag_information.out.attrib;
                break;
        case RAW_FILEINFO_STREAM_INFO: {
                uint_t c;
                DEBUG(5,("%s: async callback level RAW_FILEINFO_STREAM_INFO %s,\n",__FUNCTION__));
                d->status_RAW_FILEINFO_STREAM_INFO=status;
                DEBUG(5,("Num Streams %d %s\n",io->stream_info.out.num_streams, get_friendly_nt_error_msg (status)));
                if (NT_STATUS_IS_OK(status)) {
                        d->streams=talloc_zero_array(d, struct info_stream, io->stream_info.out.num_streams);
                        if (! d->streams) {
                                d->status_RAW_FILEINFO_STREAM_INFO=NT_STATUS_NO_MEMORY;
                        } else {
                                d->num_streams=io->stream_info.out.num_streams;
                                for(c=0; c < io->stream_info.out.num_streams; c++) {
                                        d->streams[c].size = io->stream_info.out.streams[c].size;
                                        d->streams[c].alloc_size = io->stream_info.out.streams[c].alloc_size;
                                        d->streams[c].stream_name.s=io->stream_info.out.streams[c].stream_name.s;
                                        d->streams[c].stream_name.count=io->stream_info.out.streams[c].stream_name.private_length;
                                }
                        }
                }
                break; }
        default:
                /* so... where's it from? */
                DEBUG(5,("Unexpected read level\n"));
        }

        fragment->smb_fileinfo = NULL;
        fragment->c_req=NULL;

        /* are the fragments complete? */
        DLIST_REMOVE(fragments->fragments, fragment);
        /* if this index is complete, remove from fragmentses */
        if (! fragments->fragments) {
                DLIST_REMOVE(fragmentses->fragments, fragments);
        }
        /* is that the end? */
        if (! fragmentses->fragments && fragmentses->async) {
                DEBUG(5,("Thats the end of the fragments, doing send\n"));
                /* call the send_fn */
                req=fragmentses->req;
                req->async_states->status=NT_STATUS_OK;
                DEBUG(5,("Fragments async response sending\n"));
                req->async_states->send_fn(req);
        }
        DEBUG(5,("%s: Thats the end of the callback\n",__FUNCTION__));
        return status;
}

#define FINISH_GETINFO_FRAGMENT(r, io) do { \
                struct smbcli_request *c_req; \
                switch (r->in.info_tags[0].tag_type) { \
                case TAG_TYPE_FILE_INFO: \
                        io->all_info.in.file.fnum=r->in.info_tags[0].info_tag.fnum; \
                        c_req=smb_raw_fileinfo_send(private->tree, io); \
                        break; \
                case TAG_TYPE_PATH_INFO: \
                        io->all_info.in.file.path=r->in.info_tags[0].info_tag.path.s; \
                        c_req=smb_raw_pathinfo_send(private->tree, io); \
                        break; \
                default: \
                        return NT_STATUS_INVALID_PARAMETER; \
                } \
                /* Add fragment collator */ \
                fragment->c_req=c_req; \
                /* use the same stateful async handler for them all... */ \
                { void* req=NULL; \
                ADD_ASYNC_RECV_TAIL(c_req, io, fragment, f, async_proxy_getinfo, NT_STATUS_INTERNAL_ERROR); \
                ASYNC_RECV_TAIL_HANDLER_ORPHAN(io, async_chain_handler); \
                } \
                io=NULL; \
} while (0)

#define SETUP_GETINFO_FRAGMENT(io, LEVEL) do { \
                fragment=talloc_zero(fragments, struct proxy_getinfo_fragment); \
                NT_STATUS_HAVE_NO_MEMORY(fragment); \
                DLIST_ADD(fragments->fragments, fragment); \
                fragment->fragments=fragments; \
                io=talloc_zero(fragment, union smb_fileinfo); \
                NT_STATUS_HAVE_NO_MEMORY(io); \
                io->generic.level=LEVEL; \
} while (0)

static NTSTATUS rpclite_proxy_Getinfo(struct ntvfs_module_context *ntvfs,
                                      struct ntvfs_request *req, struct proxy_GetInfo *r)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request *c_req;
        union smb_fileinfo *io=NULL;
        NTSTATUS status;
        struct proxy_file* f;
        struct ntvfs_handle *h;
        struct proxy_getinfo_fragmentses *fragmentses;
        int c;

        SETUP_PID;

        DEBUG(5,("Opnum: proxy_Getinfo r=%p\n",r));

        DEBUG(5,("Convering %d handles for r=%p\n",r->in.count, r));
        for(c=0; c < r->in.count; c++) {
                if (r->in.info_tags[c].tag_type==TAG_TYPE_FILE_INFO) {
                        RPCLITE_SETUP_THIS_FILE_HERE(r->in.info_tags[c].info_tag.fnum, f, h);
                }
        }

        if (PROXY_REMOTE_SERVER(private)) {
                DEBUG(5,("Remote proxy, doing transparent\n"));
                c_req = smbcli_ndr_request_ntioctl_send(private->tree, ntvfs, &ndr_table_rpcproxy, NDR_PROXY_GETINFO, r);
                /* No need to add a receive hander, the ntioctl transport adds
                   the async chain handler which deals with the send_fn */
//              ADD_ASYNC_RECV_TAIL(c_req, r, NULL, f, rpclite_proxy_Getinfo_map_async_send, NT_STATUS_INTERNAL_ERROR);

                if (! (req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                        DEBUG(5,("%s:Sync waiting for nttrans response\n",__LOCATION__));
                        return sync_chain_handler(c_req);
                } else {
                        ASYNC_RECV_TAIL_HANDLER_ORPHAN(io, async_chain_handler);
                        req->async_states->state |= NTVFS_ASYNC_STATE_ASYNC;
                        return NT_STATUS_OK;
                }
        }

        /* I thought this was done for me for [in,out] */
        r->out.info_data=talloc_zero_array(r, struct info_data, r->in.count);
        NT_STATUS_HAVE_NO_MEMORY(r->out.info_data);
        r->out.count = r->in.count;
        r->out.result = NT_STATUS_OK;

        fragmentses=talloc_zero(req, struct proxy_getinfo_fragmentses);
        fragmentses->r=r;
        fragmentses->req=req;
        NT_STATUS_HAVE_NO_MEMORY(fragmentses);

#warning, if C is large, we need to do a few at a time according to resource limits
        for (c=0; c < r->in.count; c++) {
                struct proxy_getinfo_fragments *fragments;
                struct proxy_getinfo_fragment *fragment;

                fragments=talloc_zero(fragmentses, struct proxy_getinfo_fragments);
                NT_STATUS_HAVE_NO_MEMORY(fragments);
                DLIST_ADD(fragmentses->fragments, fragments);
                fragments->fragmentses=fragmentses;
                fragments->index=c;

                /* Issue a set of getinfo requests */
                SETUP_GETINFO_FRAGMENT(io, RAW_FILEINFO_ALL_INFO);
                FINISH_GETINFO_FRAGMENT(r, io);

                SETUP_GETINFO_FRAGMENT(io, RAW_FILEINFO_BASIC_INFORMATION);
                FINISH_GETINFO_FRAGMENT(r, io);

                SETUP_GETINFO_FRAGMENT(io, RAW_FILEINFO_COMPRESSION_INFO);
                FINISH_GETINFO_FRAGMENT(r, io);

                SETUP_GETINFO_FRAGMENT(io, RAW_FILEINFO_INTERNAL_INFORMATION);
                FINISH_GETINFO_FRAGMENT(r, io);

                SETUP_GETINFO_FRAGMENT(io, RAW_FILEINFO_ACCESS_INFORMATION);
                FINISH_GETINFO_FRAGMENT(r, io);

                SETUP_GETINFO_FRAGMENT(io, RAW_FILEINFO_POSITION_INFORMATION);
                FINISH_GETINFO_FRAGMENT(r, io);

                SETUP_GETINFO_FRAGMENT(io, RAW_FILEINFO_MODE_INFORMATION);
                FINISH_GETINFO_FRAGMENT(r, io);

                SETUP_GETINFO_FRAGMENT(io, RAW_FILEINFO_ALIGNMENT_INFORMATION);
                FINISH_GETINFO_FRAGMENT(r, io);

                SETUP_GETINFO_FRAGMENT(io, RAW_FILEINFO_ATTRIBUTE_TAG_INFORMATION);
                FINISH_GETINFO_FRAGMENT(r, io);

                SETUP_GETINFO_FRAGMENT(io, RAW_FILEINFO_STREAM_INFO);
                FINISH_GETINFO_FRAGMENT(r, io);
        }

        /* If ! async, wait for all requests to finish */

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                struct proxy_getinfo_fragments *fragments;
                struct proxy_getinfo_fragment *fragment;
                while ((fragments = fragmentses->fragments) &&
                       (fragment = fragments->fragments) &&
                       fragment->c_req) {
                                sync_chain_handler(fragment->c_req);
                                /* and because the whole fragment / fragments may be gone now... */
                                continue;
                }
                return NT_STATUS_OK; /* see individual failures */
        }

        DEBUG(5,("%s: Setting async response\n",__FUNCTION__));
        fragmentses->async=true;
        req->async_states->state |= NTVFS_ASYNC_STATE_ASYNC;
        return NT_STATUS_OK;
}

/* rpclite dispatch table */
#define RPC_PROXY_OPS 3
struct {
    uint32_t opnum;
    NTSTATUS (*handler)(struct ntvfs_module_context *ntvfs,
                         struct ntvfs_request *req, void* r);
} rpcproxy_ops[RPC_PROXY_OPS]={
    {NDR_PROXY_READ, rpclite_proxy_Read},
    {NDR_PROXY_WRITE, rpclite_proxy_Write},
    {NDR_PROXY_GETINFO, rpclite_proxy_Getinfo}
};

/* unmarshall ntioctl and rpc-dispatch, but push async map handler to convert
   back from rpc struct to ntioctl */
static NTSTATUS proxy_rpclite(struct ntvfs_module_context *ntvfs,
                           struct ntvfs_request *req, union smb_ioctl *io)
{
        struct proxy_private *private = ntvfs->private_data;
        DATA_BLOB *request;
        struct ndr_syntax_id* syntax_id;
        uint32_t opnum;
        const struct ndr_interface_table *table;
        struct ndr_pull* pull;
        void* r;
        NTSTATUS status;
        struct async_rpclite_send *rpclite_send;
        enum ndr_err_code ndr_err;

        SETUP_PID;

        /* We don't care about io->generic.in.file, ntvfs layer already proved it was valid,
           our operations will have the fnum embedded in them anyway */
        DEBUG(5,("START %s blob-size %d\n",__FUNCTION__,io->ntioctl.in.blob.length));
        /* unpack the NDR */
        request=&io->ntioctl.in.blob;

        pull = ndr_pull_init_blob(request, req, lp_iconv_convenience(ntvfs->ctx->lp_ctx));
        NT_STATUS_HAVE_NO_MEMORY(pull);
        /* set pull->flags; LIBNDR_FLAG_PAD_CHECK, LIBNDR_FLAG_REF_ALLOC */
        DEBUG(5,("%s pull init'd\n",__FUNCTION__));

        /* the blob is 4-aligned because it was memcpy'd */
        syntax_id=talloc_zero(pull, struct ndr_syntax_id);
        NT_STATUS_HAVE_NO_MEMORY(syntax_id);

        ndr_err=ndr_pull_ndr_syntax_id(pull, NDR_SCALARS, syntax_id);
        status=ndr_map_error2ntstatus(ndr_err);
        if (! NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
                DEBUG(2,("Can't read syntax-id: %s\n",nt_errstr(status)));
                return status;
        }

        /* now find the struct ndr_interface_table * for this syntax_id */
        table=ndr_table_by_uuid(&syntax_id->uuid);
if (! table) ndr_table_init();
        table=ndr_table_by_uuid(&syntax_id->uuid);

        if (! table) {
                DEBUG(5,("Can't find table for uuid: %s\n",GUID_string(debug_ctx(),&syntax_id->uuid)));
                return NT_STATUS_NO_GUID_TRANSLATION;
        }

        ndr_err=ndr_pull_uint32(pull, NDR_SCALARS, &opnum);
        status=ndr_map_error2ntstatus(ndr_err);
        if (! NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
                DEBUG(2,("Can't read op-num: %s\n",nt_errstr(status)));
                return status;
        }
        DEBUG(5,("%s opnum %d\n",__FUNCTION__,opnum));

        DEBUG(10,("rpc request data:\n"));
        dump_data(10, pull->data, pull->data_size);

        r = talloc_named(req, table->calls[opnum].struct_size, "struct %s",
                  table->calls[opnum].name);
        NT_STATUS_HAVE_NO_MEMORY(r);

        memset(r, 0, table->calls[opnum].struct_size);

        ndr_err=table->calls[opnum].ndr_pull(pull, NDR_IN, r);
        status=ndr_map_error2ntstatus(ndr_err);
        DEBUG(5,("%s opnum %d pulled r=%p status %s\n",__FUNCTION__,opnum,r,get_friendly_nt_error_msg (status)));
        NT_STATUS_NOT_OK_RETURN(status);

        rpclite_send=talloc(req, struct async_rpclite_send);
        NT_STATUS_HAVE_NO_MEMORY(rpclite_send);
        rpclite_send->call=&table->calls[opnum];
        rpclite_send->struct_ptr=r;
        /* need to push conversion function to convert from r to io */
        status=ntvfs_map_async_setup(ntvfs, req, io, rpclite_send, proxy_rpclite_map_async_send);
        NT_STATUS_NOT_OK_RETURN(status);

        /* Magically despatch the call based on syntax_id, table and opnum.
           But there is no table of handlers.... so until then*/
        if (0==strcasecmp(table->name,"rpcproxy")) {
                if (opnum >= RPC_PROXY_OPS) {
                        DEBUG(3,("Can't despatch %s:%d\n",table->name, opnum));
                        return NT_STATUS_PROCEDURE_NOT_FOUND;
                }
                status = rpcproxy_ops[opnum].handler(ntvfs, req, r);
        } else {
                DEBUG(5,("Can't despatch %s:%d %s\n",table->name, opnum,
                                 GUID_string(debug_ctx(),&syntax_id->uuid)));
                return NT_STATUS_NO_GUID_TRANSLATION;
        }

        /* status is the status of the rpc layer. If it is NT_STATUS_OK then
           the handler status is in r->out.result */
        DEBUG(5,("%s now map_async_finish: status=%s async=%d\n", __FUNCTION__,
                 get_friendly_nt_error_msg (status), req->async_states->state & NTVFS_ASYNC_STATE_ASYNC));

        return ntvfs_map_async_finish(req, status);
}

/* unpack the ntioctl to make some rpc_struct */
NTSTATUS ntioctl_rpc_unmap(struct async_info *async, void* io1, void* io2, NTSTATUS status)
{
        struct ntvfs_module_context *ntvfs = async->proxy->ntvfs;
        struct proxy_private *proxy=async->proxy;
        struct smbcli_request *c_req = async->c_req;
        void* r=io1;
        struct ntioctl_rpc_unmap_info *info=talloc_get_type_abort(io2, struct ntioctl_rpc_unmap_info);
        union smb_ioctl* io     =talloc_get_type_abort(info->io, union smb_ioctl);
        const struct ndr_interface_call *calls=info->calls;
        enum ndr_err_code ndr_err;
        DATA_BLOB *response;
        struct ndr_pull* pull;

        DEBUG(5,("START %s io2=%p\n",__FUNCTION__,io2));
        DEBUG(5,("%s op %s ntioctl: %s\n",
                         __FUNCTION__, calls->name, get_friendly_nt_error_msg(status)));
        NT_STATUS_NOT_OK_RETURN(status);

        if (c_req) {
                DEBUG(5,("%s io2 MID=%d\n",__FUNCTION__,c_req->mid));
                status = smb_raw_ioctl_recv(c_req, io, io);
#define SESSION_INFO proxy->remote_server, proxy->remote_share
                /* This status is the ntioctl wrapper status */
                if (! NT_STATUS_IS_OK(status)) {
                        DEBUG(3,("[\\\\%s\\%s] RPC %s failed for %s: %s\n",SESSION_INFO,
                                 __FUNCTION__, calls->name, get_friendly_nt_error_msg(status)));
                        if (NT_STATUS_EQUAL(status, NT_STATUS_IO_TIMEOUT)) return status;
                        return NT_STATUS_UNSUCCESSFUL;
                }
        }

        dump_data(10, io->ntioctl.out.blob.data, io->ntioctl.out.blob.length);

        response=&io->ntioctl.out.blob;
        pull = ndr_pull_init_blob(response, r, lp_iconv_convenience(ntvfs->ctx->lp_ctx));
        /* set pull->flags; LIBNDR_FLAG_PAD_CHECK, LIBNDR_FLAG_REF_ALLOC */

        NT_STATUS_HAVE_NO_MEMORY(pull);

        ndr_err=calls->ndr_pull(pull, NDR_OUT, r);
#warning can we free pull here?
        status=ndr_map_error2ntstatus(ndr_err);

        DEBUG(5,("END %s op status %s\n",
                         __FUNCTION__, get_friendly_nt_error_msg(status)));
        return status;
}

/*
   send an ntioctl request based on a NDR encoding.
 */
struct smbcli_request *smbcli_ndr_request_ntioctl_send(
                                                struct smbcli_tree *tree,
                                                struct ntvfs_module_context *ntvfs,
                                                const struct ndr_interface_table *table,
                                                uint32_t opnum,
                                                void *r)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_request * c_req;
        struct ndr_push *push;
        NTSTATUS status;
        DATA_BLOB request;
        enum ndr_err_code ndr_err;
        union smb_ioctl *io=talloc_zero(r, union smb_ioctl);


        /* setup for a ndr_push_* call, we can't free push until the message
           actually hits the wire */
        push = ndr_push_init_ctx(io, lp_iconv_convenience(ntvfs->ctx->lp_ctx));
        if (!push) return NULL;

        /* first push interface table identifiers */
        ndr_err=ndr_push_ndr_syntax_id(push, NDR_SCALARS, &table->syntax_id);
        status=ndr_map_error2ntstatus(ndr_err);

        if (! NT_STATUS_IS_OK(status)) return NULL;

        ndr_err=ndr_push_uint32(push, NDR_SCALARS, opnum);
        status=ndr_map_error2ntstatus(ndr_err);
        if (! NT_STATUS_IS_OK(status)) return NULL;

        if (0) {
                push->flags |= LIBNDR_FLAG_BIGENDIAN;
        }

        /* push the structure into a blob */
        ndr_err = table->calls[opnum].ndr_push(push, NDR_IN, r);
        status=ndr_map_error2ntstatus(ndr_err);
        if (!NT_STATUS_IS_OK(status)) {
                DEBUG(2,("Unable to ndr_push structure in dcerpc_ndr_request_send - %s\n",
                         nt_errstr(status)));
                return NULL;
        }

        /* retrieve the blob */
        request = ndr_push_blob(push);

        io->ntioctl.level=RAW_IOCTL_NTIOCTL;
        io->ntioctl.in.function=FSCTL_UFOPROXY_RPCLITE;
        io->ntioctl.in.file.fnum=private->nttrans_fnum;
        io->ntioctl.in.fsctl=false;
        io->ntioctl.in.filter=0;
        io->ntioctl.in.max_data=PROXY_NTIOCTL_MAXDATA;
        io->ntioctl.in.blob=request;

        DEBUG(10,("smbcli_request packet:\n"));
        dump_data(10, request.data, request.length);

        c_req = smb_raw_ioctl_send(tree, io);

        if (! c_req) {
                return NULL;
        }

        dump_data(10, c_req->out.data, c_req->out.data_size);

        { void* req=NULL;
                struct ntioctl_rpc_unmap_info* info=talloc_zero(r, struct ntioctl_rpc_unmap_info);
                info->io=io;
                info->table=table;
                info->opnum=opnum;
                info->calls=&table->calls[opnum];
                ADD_ASYNC_RECV_TAIL(c_req, r, info, NULL, ntioctl_rpc_unmap, NULL);
        }

        return c_req;
}

/*
   client helpers, mapping between proxy RPC calls and smbcli_* calls.
 */

/*
 * If the sync_chain_handler is called directly it unplugs the async handler
   which (as well as preventing loops) will also avoid req->send_fn being
   called - which is also nice! */
NTSTATUS sync_chain_handler(struct smbcli_request *c_req)
{
        struct async_info *async=NULL;
        /* the first callback which will actually receive the c_req response */
        struct async_info_map *async_map;
        NTSTATUS status=NT_STATUS_OK;
        struct async_info_map** chain;

        DEBUG(5,("%s\n",__FUNCTION__));
        if (! c_req) return NT_STATUS_UNSUCCESSFUL;

        /* If there is a handler installed, it is using async_info to chain */
        if (c_req->async.fn) {
                /* not safe to talloc_free async if send_fn has been called for the request
                   against which async was allocated, so steal it (and free below) or neither */
                async = talloc_get_type_abort(c_req->async.private, struct async_info);
                talloc_steal(NULL, async);
                chain=&async->chain;
                async_map = talloc_get_type_abort(*chain, struct async_info_map);
        } else {
                chain=(struct async_info_map**)&c_req->async.private;
                async_map = talloc_get_type_abort(*chain, struct async_info_map);
        }

        /* unplug c_req->async.fn as if a callback handler calls smb_*_recv
           in order to receive the response, smbcli_transport_finish_recv will
           call us again and then call the c-req->async.fn
           Perhaps we should merely call smbcli_request_receive() IF
           c_req->request_state <= SMBCLI_REQUEST_RECV, but that might not
           help multi-part replies... except all parts are receive before
           callback if a handler WAS set */
        c_req->async.fn=NULL;

        /* Should we raise an error? Should we simple_recv? */
        while(async_map) {
                /* remove this one from the list before we call. We do this in case
                   some callbacks free their async_map but also so that callbacks
                   can navigate the async_map chain to add additional callbacks to
                   the end - e.g. so that tag-along reads can call send_fn after
                   the send_fn of the request they tagged along to, thus preserving
                   the async response order - which may be a waste of time? */
                DLIST_REMOVE(*chain, async_map);

                DEBUG(5,("Callback for async_map=%p pre-status %s\n",async_map, get_friendly_nt_error_msg(status)));
                if (async_map->fn) {
                        status=async_map->fn(async_map->async,
                                                                 async_map->parms1, async_map->parms2, status);
                }
                DEBUG(5,("Callback complete for async_map=%p status %s\n",async_map, get_friendly_nt_error_msg(status)));
                /* Note: the callback may have added to the chain */
#warning Async_maps have a null talloc_context, it is unclear who should own them
                /*  it can't be c_req as it stops us chaining more than one, maybe it
                    should be req but there isn't always a req. However sync_chain_handler
                    will always free it if called */
                DEBUG(6,("Will free async map %p\n",async_map));
#warning put me back
                talloc_free(async_map);
                DEBUG(6,("Free'd async_map\n"));
                if (*chain)
                        async_map=talloc_get_type_abort(*chain, struct async_info_map);
                else
                        async_map=NULL;
                DEBUG(6,("Switch to async_map %p\n",async_map));

                /* The first callback will have read c_req, thus talloc_free'ing it,
                   so we don't let the other callbacks get hurt playing with it */
                if (async_map && async_map->async)
                        async_map->async->c_req=NULL;
        }

        talloc_free(async);

        DEBUG(5,("%s complete: %s\n",__FUNCTION__,get_friendly_nt_error_msg (status)));
        return status;
}

/* If the async handler is called, then the send_fn is called */
static void async_chain_handler(struct smbcli_request *c_req)
{
        struct async_info *async = talloc_get_type_abort(c_req->async.private, struct async_info);
        struct ntvfs_request *req = async->req;
        NTSTATUS status;

        if (c_req->state <= SMBCLI_REQUEST_RECV) {
                /* Looks like async handlers has been called sync'ly */
                smb_panic("async_chain_handler called asyncly on req %p\n");
        }

        status=sync_chain_handler(c_req);

        /* Should we insist that a chain'd handler does this?
           Which makes it hard to intercept the data by adding handlers
           before the send_fn handler sends it... */
        if (req) {
                DEBUG(5,("%s send_fn on req=%p\n",__FUNCTION__,req));
                req->async_states->status=status;
                req->async_states->send_fn(req);
        }
}

/* unpack the rpc struct to make some smb_write */
NTSTATUS async_proxy_smb_raw_write_rpc(struct async_info *async,
                                                                           void* io1, void* io2, NTSTATUS status)
{
        union smb_write* io  =talloc_get_type(io1, union smb_write);
        struct proxy_Write* r=talloc_get_type(io2, struct proxy_Write);

        DEBUG(5,("START: %s convert from rpc to smb with pre-status %s\n",__FUNCTION__,
                         get_friendly_nt_error_msg (status)));
        DEBUG(3,("Write response for offset=%lld\n",io->generic.in.offset));
        NT_STATUS_NOT_OK_RETURN(status);

        status=r->out.result;
        DEBUG(5,("%s wrapped status: %s\n",__FUNCTION__, get_friendly_nt_error_msg(status)));
        NT_STATUS_NOT_OK_RETURN(status);

        io->generic.out.remaining = r->out.remaining;
        io->generic.out.nwritten  = r->out.nwritten;

        DEBUG(5,("END: %s status %s\n",__FUNCTION__,
                         get_friendly_nt_error_msg (status)));
        return status;
}

/* upgrade from smb to NDR and then send.
   The caller should ADD_ASYNC_RECV_TAIL the handler that tries to receive the response*/
struct smbcli_request *proxy_smb_raw_write_send(struct ntvfs_module_context *ntvfs,
                                                                                                union smb_write *io,
                                                                                                struct proxy_file *f)
{
        struct proxy_private *private = ntvfs->private_data;
        struct smbcli_tree *tree=private->tree;

        if (PROXY_REMOTE_SERVER(private)) {
                struct smbcli_request *c_req;
                struct proxy_Write *r=talloc_zero(io, struct proxy_Write);
                ssize_t size;

                if (! r) return NULL;

                size=io->generic.in.count;
                /* upgrade the write */
                r->in.fnum = io->generic.in.file.fnum;
                r->in.offset    = io->generic.in.offset;
                r->in.count     = io->generic.in.count;
                r->in.mode      = io->generic.in.wmode;
//              r->in.remaining = io->generic.in.remaining;
#warning remove this
                /* prepare to lie */
                r->out.nwritten=r->in.count;
                r->out.remaining=0;

                /* try to compress */
#warning compress!
                r->in.data.compress.data=compress_block_talloc(r, io->generic.in.data, &size);
                if (r->in.data.compress.data) {
                        r->in.data.compress.count=size;
                        r->in.flags     = PROXY_USE_ZLIB;
                } else {
                        r->in.flags     = 0;
                        /* we'll honour const, honest gov */
                        r->in.data.generic.data=discard_const(io->generic.in.data);
                        r->in.data.generic.count=io->generic.in.count;
                }

                c_req = smbcli_ndr_request_ntioctl_send(private->tree,
                                                                                                ntvfs,
                                                                                                &ndr_table_rpcproxy,
                                                                                                NDR_PROXY_WRITE, r);
                if (! c_req) return NULL;

                /* yeah, filthy abuse of f */
                { void* req=NULL;
                ADD_ASYNC_RECV_TAIL(c_req, io, r, f, async_proxy_smb_raw_write_rpc, NULL);
                }

                return c_req;
        } else {
                return smb_raw_write_send(tree, io);
        }
}

NTSTATUS proxy_smb_raw_write(struct ntvfs_module_context *ntvfs,
                                                         union smb_write *io,
                                                         struct proxy_file *f)
{
        struct proxy_private *proxy = ntvfs->private_data;
        struct smbcli_tree *tree=proxy->tree;

        if (PROXY_REMOTE_SERVER(proxy)) {
                struct smbcli_request *c_req = proxy_smb_raw_write_send(ntvfs, io, f);
                return sync_chain_handler(c_req);
        } else {
                struct smbcli_request *c_req = smb_raw_write_send(tree, io);
                return smb_raw_write_recv(c_req, io);
        }
}

/* unpack the rpc struct to make some smb_read response */
NTSTATUS async_proxy_smb_raw_read_rpc(struct async_info *async,
                                                                           void* io1, void* io2, NTSTATUS status)
{
        union smb_read* io      =talloc_get_type_abort(io1, union smb_read);
        struct proxy_Read* r=talloc_get_type_abort(io2, struct proxy_Read);
        struct proxy_file *f = async->f;
        struct proxy_private *private=async->proxy;

        DEBUG(5,("\n>>\n%s() rpc status: %s\n",__FUNCTION__,
                         get_friendly_nt_error_msg(status)));
        NT_STATUS_NOT_OK_RETURN(status);

        status=r->out.result;
        DEBUG(5,("%s() wrapped status: %s\n",__FUNCTION__,
                         get_friendly_nt_error_msg(status)));
        NT_STATUS_NOT_OK_RETURN(status);

        io->generic.out.remaining = 0; /*r->out.response.generic.remaining;*/
        io->generic.out.compaction_mode = 0;

        if (r->out.flags & (PROXY_USE_CACHE | PROXY_VALIDATE)) {
                /* Use the io we already setup!
                   if out.flags & PROXY_VALIDATE, we may need to validate more in
                   cache then r->out.nread would suggest, see io->generic.out.nread */
                if (r->out.flags & PROXY_VALIDATE)
                        io->generic.out.nread=io->generic.in.maxcnt;
                DEBUG(5,("Using cached data: size=%lld\n",
                                 (long long) io->generic.out.nread));
                return status;
        }

        if (r->in.flags & PROXY_VALIDATE) {
                DEBUG(5,("Cached data did not validate, flags: %x\n",r->out.flags));
                /* turn off validate on this file */
                //cache_handle_novalidate(f);
#warning turn off validate on this file - do an nread<maxcnt later
        }

        if (r->in.flags & PROXY_USE_CACHE) {
                DEBUG(5,("Cached data did not match\n"));
        }

        io->generic.out.nread     = r->out.nread;

        /* we may need to uncompress */
        if (r->out.flags & PROXY_USE_ZLIB) {
                ssize_t size=r->out.response.compress.count;
                DEBUG(5,("%s: uncompress, %lld wanted %lld or %lld\n",__LOCATION__,
                         (long long int)size,
                         (long long int)io->generic.in.maxcnt,
                         (long long int)io->generic.in.mincnt));
                if (size > io->generic.in.mincnt) {
                        /* we did a bulk read for the cache */
                        uint8_t *data=talloc_size(io, io->generic.in.maxcnt);
                        DEBUG(5,("%s: bulk uncompress to %p\n",__LOCATION__,data));
                        if (! uncompress_block_to(data,
                        r->out.response.compress.data, &size,
                        io->generic.in.maxcnt) ||
                        size != r->out.nread) {
                            status=NT_STATUS_INVALID_USER_BUFFER;
                        } else {
                            DEBUG(5,("%s: uncompressed\n",__LOCATION__));
                            /* copy as much as they can take */
                            io->generic.out.nread=MIN(io->generic.in.mincnt, size);
                            memcpy(io->generic.out.data, data, io->generic.out.nread);
                            /* copy the rest to the cache */
                            cache_handle_save(f, data,
                                size,
                                io->generic.in.offset);
                        }
                } else if (! uncompress_block_to(io->generic.out.data,
                        r->out.response.compress.data, &size,
                        io->generic.in.maxcnt) ||
                        size != r->out.nread) {
                        io->generic.out.nread=size;
                        status=NT_STATUS_INVALID_USER_BUFFER;
                }
        } else if (io->generic.out.data != r->out.response.generic.data) {
                //Assert(r->out.nread == r->out.generic.out.count);
                memcpy(io->generic.out.data, r->out.response.generic.data, io->generic.out.nread);
        }
        if (r->out.cache_name.s && r->out.cache_name.count && f && f->cache) {
                int result;
                setenv("WAFS_CACHE_REMOTE_NAME",r->out.cache_name.s,1);
                setenv("WAFS_CACHE_LOCAL_NAME",f->cache->cache_name,1);
                setenv("WAFS_REMOTE_SERVER",private->remote_server,1);
                DEBUG(5,("WAFS_CACHE_REMOTE_NAME=%s [cache_name]\nWAFS_CACHE_LOCAL_NAME=%s\nWAFS_REMOTE_SERVER=%s\n\n",getenv("WAFS_CACHE_REMOTE_NAME"),getenv("WAFS_CACHE_LOCAL_NAME"),getenv("WAFS_REMOTE_SERVER")));
                DEBUG(5,("%s running cache transfer command: %s\n",__LOCATION__,getenv("WAFS_CACHE_REMOTE_NAME")));
                system(getenv("WAFS_CACHE_TRANSFER"));
                DEBUG(5,("%s cache transfer command result %d\n",__LOCATION__,result));
                // now set cache to make whole local file valid
                cache_validated(f->cache, cache_len(f->cache));
        }

        return status;
}

/* Warning: Assumes that if io->generic.out.nread is not zero, then some
   data has been pre-read into io->generic.out.data and can be used for
   proxy<->proxy optimized reads */
struct smbcli_request *proxy_smb_raw_read_send(struct ntvfs_module_context *ntvfs,
                                                                                           union smb_read *io,
                                                                                           struct proxy_file *f,
                                                                                           struct proxy_Read *r)
{
        struct proxy_private *private = ntvfs->private_data;
#warning we are using out.nread as a out-of-band parameter
        if (PROXY_REMOTE_SERVER(private)) {

                struct smbcli_request *c_req;
                if (! r) {
                        r=talloc_zero(io, struct proxy_Read);
                        if (! r) return NULL;
                        r->in.mincnt    = io->generic.in.mincnt;
                }


                r->in.fnum = io->generic.in.file.fnum;
                r->in.read_for_execute=io->generic.in.read_for_execute;
                r->in.offset    = io->generic.in.offset;
                r->in.maxcnt    = io->generic.in.maxcnt;
                r->in.remaining = io->generic.in.remaining;
                r->in.flags    |= PROXY_USE_ZLIB;
                if (! (r->in.flags & PROXY_VALIDATE) &&
                        io->generic.out.data && io->generic.out.nread > 0) {
                        /* maybe we should limit digest size to MIN(nread, maxcnt) to
                           permit the caller to provider a larger nread as part of
                           a split read */
                        checksum_block(r->in.digest.digest, io->generic.out.data,
                                                   io->generic.out.nread);

                        if (io->generic.out.nread > r->in.maxcnt) {
                                DEBUG(0,("Cache from nread is too big for requested read struct, ignoring cache\n"));
                        } else {
                                r->in.mincnt = io->generic.out.nread;
                                r->in.maxcnt = io->generic.out.nread;
                                r->in.flags |= PROXY_USE_CACHE;
                                /* PROXY_VALIDATE will have been set by caller */
                        }
                }

                if (r->in.flags & (PROXY_USE_CACHE | PROXY_VALIDATE)) {
                        DEBUG(3,("Cache digest length=%lld\n", (long long)r->in.maxcnt));
                        dump_data (3, r->in.digest.digest, sizeof(r->in.digest.digest));
                }

                c_req = smbcli_ndr_request_ntioctl_send(private->tree,
                                                                                                ntvfs,
                                                                                                &ndr_table_rpcproxy,
                                                                                                NDR_PROXY_READ, r);
                if (! c_req) return NULL;

                { void* req=NULL;
                ADD_ASYNC_RECV_TAIL(c_req, io, r, f, async_proxy_smb_raw_read_rpc, NULL);
                }

                return c_req;
        } else {
                return smb_raw_read_send(private->tree, io);
        }
}

NTSTATUS proxy_smb_raw_read(struct ntvfs_module_context *ntvfs,
                                                        union smb_read *io,
                                                        struct proxy_file *f)
{
        struct proxy_private *proxy = ntvfs->private_data;
        struct smbcli_tree *tree=proxy->tree;

        if (PROXY_REMOTE_SERVER(proxy)) {
                struct smbcli_request *c_req = proxy_smb_raw_read_send(ntvfs, io, f, NULL);
                return sync_chain_handler(c_req);
        } else {
                struct smbcli_request *c_req = smb_raw_read_send(tree, io);
                return smb_raw_read_recv(c_req, io);
        }
}


/*
  initialise the PROXY->PROXY backend, registering ourselves with the ntvfs subsystem
 */
NTSTATUS ntvfs_proxy_init(void)
{
        NTSTATUS ret;
        struct ntvfs_ops ops;
        NTVFS_CURRENT_CRITICAL_SIZES(vers);

        ZERO_STRUCT(ops);

        /* fill in the name and type */
        ops.name = "proxy";
        ops.type = NTVFS_DISK;

        /* fill in all the operations */
        ops.connect = proxy_connect;
        ops.disconnect = proxy_disconnect;
        ops.unlink = proxy_unlink;
        ops.chkpath = proxy_chkpath;
        ops.qpathinfo = proxy_qpathinfo;
        ops.setpathinfo = proxy_setpathinfo;
        ops.open = proxy_open;
        ops.mkdir = proxy_mkdir;
        ops.rmdir = proxy_rmdir;
        ops.rename = proxy_rename;
        ops.copy = proxy_copy;
        ops.ioctl = proxy_ioctl;
        ops.read = proxy_read;
        ops.write = proxy_write;
        ops.seek = proxy_seek;
        ops.flush = proxy_flush;
        ops.close = proxy_close;
        ops.exit = proxy_exit;
        ops.lock = proxy_lock;
        ops.setfileinfo = proxy_setfileinfo;
        ops.qfileinfo = proxy_qfileinfo;
        ops.fsinfo = proxy_fsinfo;
        ops.lpq = proxy_lpq;
        ops.search_first = proxy_search_first;
        ops.search_next = proxy_search_next;
        ops.search_close = proxy_search_close;
        ops.trans = proxy_trans;
        ops.logoff = proxy_logoff;
        ops.async_setup = proxy_async_setup;
        ops.cancel = proxy_cancel;
        ops.notify = proxy_notify;
        ops.trans2 = proxy_trans2;

        /* register ourselves with the NTVFS subsystem. We register
           under the name 'proxy'. */
        ret = ntvfs_register(&ops, &vers);

        if (!NT_STATUS_IS_OK(ret)) {
                DEBUG(0,("Failed to register PROXY backend!\n"));
        }

        return ret;
}