Fix broken oplock handling

[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.

*/

#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 PROXY_NTIOCTL_MAXDATA 0x20000

#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"

struct proxy_file {
        struct proxy_file *prev, *next;
        uint16_t fnum;
        struct ntvfs_handle *h;
        struct cache_file_entry *cache;
        int readahead_pending;
        int oplock;
};

/* 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;
        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;
        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 */
};

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 { \
        if ((h = ntvfs_find_handle(private->ntvfs, req, r->in.fnum)) && \
            (f = ntvfs_handle_get_backend_data(h, ntvfs))) { \
            r->in.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))

/* 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);

/*
  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;

        for (f=private->files; f; f=f->next) {
                if (f->fnum != fnum) continue;
                h = f->h;
                break;
        }

        if (!h) {
                DEBUG(5,("vfs_proxy: ignoring oplock break level %d for fnum %d\n", level, fnum));
                return true;
        }

        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)) return false;
        return true;
}

/*
  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);
                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, "\\",
                                                    NTCREATEX_FLAGS_OPEN_DIRECTORY,
                                                    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;

        /* 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); \
        TYPE_CHECK(void (*t)(struct smbcli_request *),async_fn); \
        { \
                struct async_info *async; \
                async = talloc(req, struct async_info); \
                if (!async) return (error); \
                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) return (error); \
        { \
                struct async_info_map *async_map=talloc(NULL, struct async_info_map); \
                if (! async_map) return (error); \
                async_map->async=talloc(async_map, struct async_info); \
                if (! async_map->async) 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 */ \
                        return (error); \
                } else { \
                        DLIST_ADD_END(creq->async.private, async_map, struct async_info_map *); \
                } \
        } \
} while(0)

/* 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))

/*
  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;
}

/*
  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);
}

/*
  return info on a pathname
*/
static NTSTATUS proxy_qpathinfo(struct ntvfs_module_context *ntvfs,
                               struct ntvfs_request *req, union smb_fileinfo *info)
{
        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_pathinfo(private->tree, req, info);
        }

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

        ASYNC_RECV_TAIL(info, 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);
}

/*
  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;

        SETUP_PID_AND_FILE;

        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;

        if (proxy->cache_enabled) {
                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);
        }

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;
        NTSTATUS status;

        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->h = h;

        /* 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;

                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 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,("\n\n%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;
        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;
        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\n",__FUNCTION__,f->fnum));

        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;
        MD5Final(parts->digest, &parts->context);

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

        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 \n\n\n"));
                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",
                                __FUNCTION__,r->out.nread,size,size*100/r->out.nread));
                }
        }

        /* 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;

        DEBUG(5,("%s: parts=%p c_req=%p io_frag=%p\n",__FUNCTION__,parts, c_req, io_frag));
        /* 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)));

        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 */
                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, r->in.offset);
                /* Don't want to go past mincnt */
                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) */
                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;
                }

                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;

        //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",__FUNCTION__,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,("%s offset=%lld, mincnt=%d, maxcnt=%d\n",__FUNCTION__,
                  io->generic.in.offset,
                  io->generic.in.mincnt,
                  io->generic.in.maxcnt));
        io->generic.out.nread=0;
        /* 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;
                        }
                }
        }

        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;
                                        /* the proxy send function will calculate the checksum based on *data */
                                } else {
                                        /* 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 (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);
        }

                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;

        SETUP_PID;

        if (io->generic.level != RAW_CLOSE_GENERIC &&
            private->map_generic) {
                return ntvfs_map_close(ntvfs, req, io);
        }
        SETUP_FILE_HERE(f);
        /* 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);

        /* 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);
        }

        if (!(req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
                return smbcli_request_simple_recv(c_req);
        }

        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;
        req->async_states->status = smb_raw_fsinfo_recv(c_req, req, async->parms);
        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;

        /* 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)) {
                return smb_raw_fsinfo(private->tree, req, fs);
        }

        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;
}

/*
   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;

        SETUP_PID;

        return smb_raw_search_first(private->tree, req, io, search_private, callback);
}

/* 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;

        SETUP_PID;

        return smb_raw_search_next(private->tree, req, io, search_private, callback);
}

/* 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;

        SETUP_PID;

        return smb_raw_search_close(private->tree, io);
}

/*
  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);

        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);
}

/* 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) {
                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);

        ndr_err=table->calls[opnum].ndr_pull(pull, NDR_IN, r);
        status=ndr_map_error2ntstatus(ndr_err);
        DEBUG(5,("%s opnum %d pulled status %s\n",__FUNCTION__,opnum,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);

        /* 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")) {
                switch(opnum) {
                        case(NDR_PROXY_READ):
                                status=rpclite_proxy_Read(ntvfs, req, r);
                                break;
                        case(NDR_PROXY_WRITE):
                                status=rpclite_proxy_Write(ntvfs, req, r);
                                break;
                        default:
                                DEBUG(3,("Can't despatch %s:%d\n",table->name, opnum));
                                return NT_STATUS_PROCEDURE_NOT_FOUND;
                }
        } 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 */
        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) {
                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);

        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;
                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);
        }

        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.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.mincnt    = io->generic.in.mincnt;
                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;
}