src/doc/html/ipcalc_c.html


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381

<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.0 Transitional//EN">
<html><head><meta name="robots" content="noindex">
<meta http-equiv="Content-Type" content="text/html;charset=iso-8859-1">
<title>ipcalc.c File Reference</title>
<link href="doxygen.css" rel="stylesheet" type="text/css">
</head><body bgcolor="#ffffff">
<!-- Generated by Doxygen 1.2.6 -->
<center>
<a class="qindex" href="index.html">Main Page</a> &nbsp; <a class="qindex" href="files.html">File List</a> &nbsp; <a class="qindex" href="globals.html">Globals</a> &nbsp; </center>
<hr><h1>ipcalc.c File Reference</h1>provides utilities for manipulating IP addresses. 
<a href="#_details">More...</a>
<p>
<code>#include &lt;ctype.h&gt;</code><br>
<code>#include &lt;popt.h&gt;</code><br>
<code>#include &lt;stdio.h&gt;</code><br>
<code>#include &lt;stdlib.h&gt;</code><br>
<code>#include &lt;string.h&gt;</code><br>
<code>#include &lt;sys/socket.h&gt;</code><br>
<code>#include &lt;netinet/in.h&gt;</code><br>
<code>#include &lt;arpa/inet.h&gt;</code><br>
<code>#include &lt;netdb.h&gt;</code><br>
<table border=0 cellpadding=0 cellspacing=0>
<tr><td colspan=2><br><h2>Defines</h2></td></tr>
<tr><td nowrap align=right valign=top><a name="a0" doxytag="ipcalc.c::IPBITS"></a>
#define&nbsp;</td><td valign=bottom><a class="el" href="ipcalc_c.html#a0">IPBITS</a>&nbsp;&nbsp;&nbsp;(sizeof(unsigned long int) * 8)</td></tr>
<tr><td>&nbsp;</td><td><font size=-1><em>the number of bits in an IP address.</em></font><br><br></td></tr>
<tr><td nowrap align=right valign=top><a name="a1" doxytag="ipcalc.c::IPBYTES"></a>
#define&nbsp;</td><td valign=bottom><a class="el" href="ipcalc_c.html#a1">IPBYTES</a>&nbsp;&nbsp;&nbsp;(sizeof(unsigned long int))</td></tr>
<tr><td>&nbsp;</td><td><font size=-1><em>the number of bytes in an IP address.</em></font><br><br></td></tr>
<tr><td colspan=2><br><h2>Functions</h2></td></tr>
<tr><td nowrap align=right valign=top>unsigned long int&nbsp;</td><td valign=bottom><a class="el" href="ipcalc_c.html#a2">prefix2mask</a> (int prefix)</td></tr>
<tr><td>&nbsp;</td><td><font size=-1><em>creates a netmask from a specified number of bits.</em> <a href="#a2">More...</a><em></em></font><br><br></td></tr>
<tr><td nowrap align=right valign=top>int&nbsp;</td><td valign=bottom><a class="el" href="ipcalc_c.html#a3">mask2prefix</a> (unsigned long int mask)</td></tr>
<tr><td>&nbsp;</td><td><font size=-1><em>calculates the number of bits masked off by a netmask.</em> <a href="#a3">More...</a><em></em></font><br><br></td></tr>
<tr><td nowrap align=right valign=top>unsigned long int&nbsp;</td><td valign=bottom><a class="el" href="ipcalc_c.html#a4">default_netmask</a> (unsigned long int addr)</td></tr>
<tr><td>&nbsp;</td><td><font size=-1><em>returns the default (canonical) netmask associated with specified IP address.</em> <a href="#a4">More...</a><em></em></font><br><br></td></tr>
<tr><td nowrap align=right valign=top>unsigned long int&nbsp;</td><td valign=bottom><a class="el" href="ipcalc_c.html#a5">calc_broadcast</a> (unsigned long int addr, int prefix)</td></tr>
<tr><td>&nbsp;</td><td><font size=-1><em>calculate broadcast address given an IP address and a prefix length.</em> <a href="#a5">More...</a><em></em></font><br><br></td></tr>
<tr><td nowrap align=right valign=top>unsigned long int&nbsp;</td><td valign=bottom><a class="el" href="ipcalc_c.html#a6">calc_network</a> (unsigned long int addr, int prefix)</td></tr>
<tr><td>&nbsp;</td><td><font size=-1><em>calculates the network address for a specified address and prefix.</em> <a href="#a6">More...</a><em></em></font><br><br></td></tr>
<tr><td nowrap align=right valign=top>const char*&nbsp;</td><td valign=bottom><a class="el" href="ipcalc_c.html#a7">get_hostname</a> (unsigned long int addr)</td></tr>
<tr><td>&nbsp;</td><td><font size=-1><em>returns the hostname associated with the specified IP address.</em> <a href="#a7">More...</a><em></em></font><br><br></td></tr>
<tr><td nowrap align=right valign=top>int&nbsp;</td><td valign=bottom><a class="el" href="ipcalc_c.html#a8">main</a> (int argc, const char **argv)</td></tr>
<tr><td>&nbsp;</td><td><font size=-1><em>wrapper program for ipcalc functions.</em> <a href="#a8">More...</a><em></em></font><br><br></td></tr>
</table>
<hr><a name="_details"></a><h2>Detailed Description</h2>
provides utilities for manipulating IP addresses.
<p>

<p>

<p>
 ipcalc provides utilities and a front-end command line interface for manipulating IP addresses, and calculating various aspects of an ip address/netmask/network address/prefix/etc.
<p>
Functionality can be accessed from other languages from the library interface, documented here. To use ipcalc from the shell, read the ipcalc(1) manual page.
<p>
When passing parameters to the various functions, take note of whether they take host byte order or network byte order. Most take host byte order, and return host byte order, but there are some exceptions.
<p>
<hr><h2>Function Documentation</h2>
<a name="a5" doxytag="ipcalc.c::calc_broadcast"></a><p>
<table width="100%" cellpadding="2" cellspacing="0" border="0">
  <tr>
    <td class="md">
      <table cellpadding="0" cellspacing="0" border="0">
        <tr>
          <td nowrap valign="top"><b> 
unsigned long int calc_broadcast (
          </b></td>
          <td valign="bottom"><b>
unsigned long int <em>addr</em>, 
          </b></td>
        </tr>
        <tr>
          <td></td>
          <td><b>
int <em>prefix</em>&nbsp;)
          </b></td>
        </tr>

      </table>
    </td>
  </tr>
</table>
<table cellspacing=5 cellpadding=0 border=0>
  <tr>
    <td>
      &nbsp;
    </td>
    <td>

<p>
calculate broadcast address given an IP address and a prefix length.
<p>

<p>
 <dl compact><dt>
<b>Parameters: </b><dd>
<table border=0 cellspacing=2 cellpadding=0>
<tr><td valign=top><em>addr</em>
&nbsp;</td><td>
an IP address in network byte order. </td></tr>
<tr><td valign=top><em>prefix</em>
&nbsp;</td><td>
a prefix length.
<p>
</td></tr>
</table>
</dl><dl compact><dt>
<b>Returns: </b><dd>
the calculated broadcast address for the network, in network byte order. </dl>    </td>
  </tr>
</table>
<a name="a6" doxytag="ipcalc.c::calc_network"></a><p>
<table width="100%" cellpadding="2" cellspacing="0" border="0">
  <tr>
    <td class="md">
      <table cellpadding="0" cellspacing="0" border="0">
        <tr>
          <td nowrap valign="top"><b> 
unsigned long int calc_network (
          </b></td>
          <td valign="bottom"><b>
unsigned long int <em>addr</em>, 
          </b></td>
        </tr>
        <tr>
          <td></td>
          <td><b>
int <em>prefix</em>&nbsp;)
          </b></td>
        </tr>

      </table>
    </td>
  </tr>
</table>
<table cellspacing=5 cellpadding=0 border=0>
  <tr>
    <td>
      &nbsp;
    </td>
    <td>

<p>
calculates the network address for a specified address and prefix.
<p>

<p>
 <dl compact><dt>
<b>Parameters: </b><dd>
<table border=0 cellspacing=2 cellpadding=0>
<tr><td valign=top><em>addr</em>
&nbsp;</td><td>
an IP address, in network byte order </td></tr>
<tr><td valign=top><em>prefix</em>
&nbsp;</td><td>
the network prefix </td></tr>
</table>
</dl><dl compact><dt>
<b>Returns: </b><dd>
the base address of the network that addr is associated with, in network byte order. </dl>    </td>
  </tr>
</table>
<a name="a4" doxytag="ipcalc.c::default_netmask"></a><p>
<table width="100%" cellpadding="2" cellspacing="0" border="0">
  <tr>
    <td class="md">
      <table cellpadding="0" cellspacing="0" border="0">
        <tr>
          <td nowrap valign="top"><b> 
unsigned long int default_netmask (
          </b></td>
          <td valign="bottom"><b>
unsigned long int <em>addr</em>&nbsp;)
          </b></td>
        </tr>

      </table>
    </td>
  </tr>
</table>
<table cellspacing=5 cellpadding=0 border=0>
  <tr>
    <td>
      &nbsp;
    </td>
    <td>

<p>
returns the default (canonical) netmask associated with specified IP address.
<p>

<p>
 When the Internet was originally set up, various ranges of IP addresses were segmented into three network classes: A, B, and C. This function will return a netmask that is associated with the IP address specified defining where it falls in the predefined classes.<dl compact><dt>
<b>Parameters: </b><dd>
<table border=0 cellspacing=2 cellpadding=0>
<tr><td valign=top><em>addr</em>
&nbsp;</td><td>
an IP address in network byte order. </td></tr>
</table>
</dl><dl compact><dt>
<b>Returns: </b><dd>
a netmask in network byte order. </dl>    </td>
  </tr>
</table>
<a name="a7" doxytag="ipcalc.c::get_hostname"></a><p>
<table width="100%" cellpadding="2" cellspacing="0" border="0">
  <tr>
    <td class="md">
      <table cellpadding="0" cellspacing="0" border="0">
        <tr>
          <td nowrap valign="top"><b> 
const char * get_hostname (
          </b></td>
          <td valign="bottom"><b>
unsigned long int <em>addr</em>&nbsp;)
          </b></td>
        </tr>

      </table>
    </td>
  </tr>
</table>
<table cellspacing=5 cellpadding=0 border=0>
  <tr>
    <td>
      &nbsp;
    </td>
    <td>

<p>
returns the hostname associated with the specified IP address.
<p>

<p>
 <dl compact><dt>
<b>Parameters: </b><dd>
<table border=0 cellspacing=2 cellpadding=0>
<tr><td valign=top><em>addr</em>
&nbsp;</td><td>
an IP address to find a hostname for, in network byte order
<p>
</td></tr>
</table>
</dl><dl compact><dt>
<b>Returns: </b><dd>
a hostname, or NULL if one cannot be determined. Hostname is stored in a static buffer that may disappear at any time, the caller should copy the data if it needs permanent storage. </dl>    </td>
  </tr>
</table>
<a name="a8" doxytag="ipcalc.c::main"></a><p>
<table width="100%" cellpadding="2" cellspacing="0" border="0">
  <tr>
    <td class="md">
      <table cellpadding="0" cellspacing="0" border="0">
        <tr>
          <td nowrap valign="top"><b> 
main (
          </b></td>
          <td valign="bottom"><b>
int <em>argc</em>, 
          </b></td>
        </tr>
        <tr>
          <td></td>
          <td><b>
const char ** <em>argv</em>&nbsp;)
          </b></td>
        </tr>

      </table>
    </td>
  </tr>
</table>
<table cellspacing=5 cellpadding=0 border=0>
  <tr>
    <td>
      &nbsp;
    </td>
    <td>

<p>
wrapper program for ipcalc functions.
<p>

<p>
 This is a wrapper program for the functions that the ipcalc library provides. It can be used from shell scripts or directly from the command line.
<p>
For more information, please see the ipcalc(1) man page.     </td>
  </tr>
</table>
<a name="a3" doxytag="ipcalc.c::mask2prefix"></a><p>
<table width="100%" cellpadding="2" cellspacing="0" border="0">
  <tr>
    <td class="md">
      <table cellpadding="0" cellspacing="0" border="0">
        <tr>
          <td nowrap valign="top"><b> 
int mask2prefix (
          </b></td>
          <td valign="bottom"><b>
unsigned long int <em>mask</em>&nbsp;)
          </b></td>
        </tr>

      </table>
    </td>
  </tr>
</table>
<table cellspacing=5 cellpadding=0 border=0>
  <tr>
    <td>
      &nbsp;
    </td>
    <td>

<p>
calculates the number of bits masked off by a netmask.
<p>

<p>
 This function calculates the significant bits in an IP address as specified by a netmask. See also <a class="el" href="ipcalc_c.html#a2">prefix2mask</a>.<dl compact><dt>
<b>Parameters: </b><dd>
<table border=0 cellspacing=2 cellpadding=0>
<tr><td valign=top><em>mask</em>
&nbsp;</td><td>
is the netmask, specified as an unsigned long integer in network byte order. </td></tr>
</table>
</dl><dl compact><dt>
<b>Returns: </b><dd>
the number of significant bits. </dl>    </td>
  </tr>
</table>
<a name="a2" doxytag="ipcalc.c::prefix2mask"></a><p>
<table width="100%" cellpadding="2" cellspacing="0" border="0">
  <tr>
    <td class="md">
      <table cellpadding="0" cellspacing="0" border="0">
        <tr>
          <td nowrap valign="top"><b> 
unsigned long int prefix2mask (
          </b></td>
          <td valign="bottom"><b>
int <em>bits</em>&nbsp;)
          </b></td>
        </tr>

      </table>
    </td>
  </tr>
</table>
<table cellspacing=5 cellpadding=0 border=0>
  <tr>
    <td>
      &nbsp;
    </td>
    <td>

<p>
creates a netmask from a specified number of bits.
<p>

<p>
 This function converts a prefix length to a netmask. As CIDR (classless internet domain internet domain routing) has taken off, more an more IP addresses are being specified in the format address/prefix (i.e. 192.168.2.3/24, with a corresponding netmask 255.255.255.0). If you need to see what netmask corresponds to the prefix part of the address, this is the function. See also <a class="el" href="ipcalc_c.html#a3">mask2prefix</a>.<dl compact><dt>
<b>Parameters: </b><dd>
<table border=0 cellspacing=2 cellpadding=0>
<tr><td valign=top><em>prefix</em>
&nbsp;</td><td>
is the number of bits to create a mask for. </td></tr>
</table>
</dl><dl compact><dt>
<b>Returns: </b><dd>
a network mask, in network byte order. </dl>    </td>
  </tr>
</table>
<hr><address><small>Generated at Mon Apr 30 16:39:02 2001 for initscripts by
<a href="http://www.doxygen.org/index.html">
<img src="doxygen.gif" alt="doxygen" align="middle" border=0 
width=110 height=53></a>1.2.6 written by <a href="mailto:dimitri@stack.nl">Dimitri van Heesch</a>,
 &copy;&nbsp;1997-2001</small></address>
</body>
</html>
/*
  Copyright (c) 2009-2011 Gluster, Inc. <http://www.gluster.com>
  This file is part of GlusterFS.

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

  GlusterFS 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/>.
*/

#include "stat-prefetch.h"
#include "statedump.h"

#define GF_SP_CACHE_BUCKETS 1
#define GF_SP_CACHE_ENTRIES_EXPECTED (128 * 1024) //1048576

typedef enum {
        SP_EXPECT,
        SP_DONT_EXPECT,
        SP_DONT_CARE
} sp_expect_t;


void
sp_inode_ctx_free (xlator_t *this, sp_inode_ctx_t *ctx)
{
        call_stub_t *stub = NULL, *tmp = NULL;

        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, out);
        GF_VALIDATE_OR_GOTO (this->name, ctx, out);

        LOCK (&ctx->lock);
        {
                if (!list_empty (&ctx->waiting_ops)) {
                        gf_log (this->name, GF_LOG_WARNING, "inode ctx is "
                                "being freed even when there are file "
                                "operations waiting for lookup-behind to "
                                "complete. The operations in the waiting list "
                                "are:");
                        list_for_each_entry_safe (stub, tmp, &ctx->waiting_ops,
                                                  list) {
                                gf_log (this->name, GF_LOG_WARNING,
                                        "OP (%s)", gf_fop_list[stub->fop]);

                                list_del_init (&stub->list);
                                call_stub_destroy (stub);
                        }
                }
        }
        UNLOCK (&ctx->lock);

        LOCK_DESTROY (&ctx->lock);
        GF_FREE (ctx);

out:
        return;
}


sp_inode_ctx_t *
sp_inode_ctx_init ()
{
        sp_inode_ctx_t *inode_ctx = NULL;

        inode_ctx = GF_CALLOC (1, sizeof (*inode_ctx), gf_sp_mt_sp_inode_ctx_t);
        if (inode_ctx == NULL) {
                goto out;
        }

        LOCK_INIT (&inode_ctx->lock);
        INIT_LIST_HEAD (&inode_ctx->waiting_ops);

out:
        return inode_ctx;
}


int
sp_update_inode_ctx (xlator_t *this, inode_t *inode, int32_t *op_ret,
                     int32_t *op_errno, char *lookup_in_progress,
                     char *looked_up, struct iatt *stbuf,
                     struct list_head *waiting_ops, int32_t *error)
{
        int32_t         ret       = -1;
        sp_inode_ctx_t *inode_ctx = NULL;
        uint64_t        value     = 0;

        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, out);
        GF_VALIDATE_OR_GOTO (this->name, inode, out);

        ret = inode_ctx_get (inode, this, &value);
        if (ret == 0) {
                inode_ctx = (sp_inode_ctx_t *)(long)value;
        }

        if (inode_ctx == NULL) {
                ret = -1;
                if (error != NULL) {
                        *error = EINVAL;
                }

                gf_log (this->name, GF_LOG_WARNING,
                        "stat-prefetch context not set in inode "
                        "(ino:%"PRId64" gfid:%s)", inode->ino,
                        uuid_utoa (inode->gfid));
                goto out;
        }

        LOCK (&inode_ctx->lock);
        {
                if (op_ret != NULL) {
                        inode_ctx->op_ret = *op_ret;
                }

                if (op_errno != NULL) {
                        inode_ctx->op_errno = *op_errno;
                }

                if (looked_up != NULL) {
                        inode_ctx->looked_up = *looked_up;
                }

                if (lookup_in_progress != NULL) {
                        inode_ctx->lookup_in_progress = *lookup_in_progress;
                }

                if ((op_ret != NULL ) && (*op_ret == 0) && (stbuf != NULL)
                    && IA_ISDIR (stbuf->ia_type)) {
                        memcpy (&inode_ctx->stbuf, stbuf, sizeof (*stbuf));
                }

                if (waiting_ops != NULL) {
                        list_splice_init (&inode_ctx->waiting_ops, waiting_ops);
                }
        }
        UNLOCK (&inode_ctx->lock);

        ret = 0;

out:
        return ret;
}


sp_inode_ctx_t *
sp_check_and_create_inode_ctx (xlator_t *this, inode_t *inode,
                               sp_expect_t expect)
{
        uint64_t        value     = 0;
        sp_inode_ctx_t *inode_ctx = NULL;
        int32_t         ret       = 0;

        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, out);
        GF_VALIDATE_OR_GOTO (this->name, inode, out);

        LOCK (&inode->lock);
        {
                ret = __inode_ctx_get (inode, this, &value);
                if (ret == 0) {
                        inode_ctx = (sp_inode_ctx_t *)(long)value;

                        if ((expect == SP_DONT_EXPECT) && (inode_ctx != NULL)) {
                                gf_log_callingfn (this->name, GF_LOG_WARNING,
                                                  "stat-prefetch context is "
                                                  "present in inode "
                                                  "(ino:%"PRId64" gfid:%s) "
                                                  "when it is supposed to be "
                                                  "not present", inode->ino,
                                                  uuid_utoa (inode->gfid));
                        }
                } else {
                        if (expect == SP_EXPECT) {
                                gf_log_callingfn (this->name, GF_LOG_WARNING,
                                                  "stat-prefetch context is "
                                                  "not present in inode "
                                                  "(ino:%"PRId64" gfid:%s)"
                                                  " when it is supposed to be "
                                                  "present", inode->ino,
                                                  uuid_utoa (inode->gfid));
                        }

                        inode_ctx = sp_inode_ctx_init ();
                        if (inode_ctx != NULL) {
                                ret = __inode_ctx_put (inode, this,
                                                       (long)inode_ctx);
                                if (ret == -1) {
                                        sp_inode_ctx_free (this, inode_ctx);
                                        inode_ctx = NULL;
                                }
                        }
                }
        }
        UNLOCK (&inode->lock);

out:
        return inode_ctx;
}


sp_cache_t *
sp_cache_ref (sp_cache_t *cache)
{
        if (cache == NULL) {
                goto out;
        }

        LOCK (&cache->lock);
        {
                cache->ref++;
        }
        UNLOCK (&cache->lock);

out:
        return cache;;
}


void
sp_cache_unref (sp_cache_t *cache)
{
        int refcount = 0;

        if (cache == NULL) {
                goto out;
        }

        LOCK (&cache->lock);
        {
                refcount = --cache->ref;
        }
        UNLOCK (&cache->lock);

        if (refcount == 0) {
                rbthash_table_destroy (cache->table);
                GF_FREE (cache);
        }

out:
        return;
}


int32_t
sp_process_inode_ctx (call_frame_t *frame, xlator_t *this, loc_t *loc,
                      call_stub_t *stub, char *need_unwind, char *need_lookup,
                      char *can_wind, int32_t *error)
{
        int32_t         ret          = -1, op_errno = EINVAL;
        sp_local_t     *local        = NULL;
        sp_inode_ctx_t *inode_ctx    = NULL;
        uint64_t        value        = 0;

        if (need_unwind != NULL) {
                *need_unwind = 1;
        }

        GF_VALIDATE_OR_GOTO ("stat-prefetch", frame, out);
        GF_VALIDATE_OR_GOTO (frame->this->name, this, out);
        GF_VALIDATE_OR_GOTO (frame->this->name, loc, out);
        GF_VALIDATE_OR_GOTO (frame->this->name, loc->inode, out);
        GF_VALIDATE_OR_GOTO (frame->this->name, need_unwind, out);
        GF_VALIDATE_OR_GOTO (frame->this->name, need_lookup, out);
        GF_VALIDATE_OR_GOTO (frame->this->name, can_wind, out);

        inode_ctx_get (loc->inode, this, &value);

        inode_ctx = (sp_inode_ctx_t *)(long) value;
        if (inode_ctx == NULL) {
                gf_log_callingfn (this->name, GF_LOG_WARNING,
                                  "stat-prefetch context not set in inode "
                                  "(ino:%"PRId64" gfid:%s)", loc->inode->ino,
                                  uuid_utoa (loc->inode->gfid));
                *can_wind = 1;
                *need_unwind = 0;
                op_errno = 0;
                ret = 0;
                goto out;
        }

        LOCK (&inode_ctx->lock);
        {
                if (!(inode_ctx->looked_up || inode_ctx->lookup_in_progress)) {
                        if (frame->local == NULL) {
                                local = GF_CALLOC (1, sizeof (*local),
                                                   gf_sp_mt_sp_local_t);
                                if (local == NULL) {
                                        op_errno = ENOMEM;
                                        goto unlock;
                                }

                                frame->local = local;

                                ret = loc_copy (&local->loc, loc);
                                if (ret == -1) {
                                        op_errno = errno;
                                        gf_log (this->name, GF_LOG_WARNING,
                                                "loc_copy failed (%s)",
                                                strerror (op_errno));
                                        goto unlock;
                                }
                        }

                        *need_lookup = 1;
                        inode_ctx->lookup_in_progress = 1;
                }

                if (inode_ctx->looked_up) {
                        *can_wind = 1;
                } else {
                        list_add_tail (&stub->list, &inode_ctx->waiting_ops);
                        stub = NULL;
                }

                *need_unwind = 0;
                ret = 0;
        }
unlock:
        UNLOCK (&inode_ctx->lock);

out:
        if (stub != NULL) {
                call_stub_destroy (stub);
        }

        if (error != NULL) {
                *error = op_errno;
        }

        return ret;
}


inline uint32_t
sp_hashfn (void *data, int len)
{
        return gf_dm_hashfn ((const char *)data, len);
}


sp_cache_t *
sp_cache_init (xlator_t *this)
{
        sp_cache_t      *cache = NULL;
        sp_private_t    *priv  = NULL;

        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, out);

        priv = this->private;
        GF_VALIDATE_OR_GOTO (this->name, priv, out);
        GF_VALIDATE_OR_GOTO (this->name, priv->mem_pool, out);

        cache = GF_CALLOC (1, sizeof (*cache), gf_sp_mt_sp_cache_t);
        if (cache) {
                cache->table =
                        rbthash_table_init (GF_SP_CACHE_BUCKETS,
                                            sp_hashfn, __gf_free,
                                            0, priv->mem_pool);
                if (cache->table == NULL) {
                        gf_log (this->name, GF_LOG_WARNING,
                                "cannot init a new rbthash table to hold "
                                "cache");
                        GF_FREE (cache);
                        cache = NULL;
                        goto out;
                }

                LOCK_INIT (&cache->lock);
                cache->this = this;
        }

out:
        return cache;
}


void
sp_local_free (sp_local_t *local)
{
        if (local) {
                loc_wipe (&local->loc);
                GF_FREE (local);
        }
}


int32_t
sp_cache_remove_entry (sp_cache_t *cache, char *name, char remove_all)
{
        int32_t          ret   = -1;
        rbthash_table_t *table = NULL;
        xlator_t        *this  = NULL;
        sp_private_t    *priv  = NULL;
        void            *data  = NULL;

        GF_VALIDATE_OR_GOTO ("stat-prefetch", cache, out);
        if ((name == NULL) && !remove_all) {
                gf_log ((cache->this ? cache->this->name : "stat-prefetch"),
                        GF_LOG_WARNING,
                        "request to remove a single entry from cache and is no "
                        "name passed to identify it");
                goto out;
        }

        this = cache->this;
        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, out);
        GF_VALIDATE_OR_GOTO (this->name, this->private, out);

        priv = this->private;

        LOCK (&cache->lock);
        {
                if (remove_all) {
                        table = cache->table;
                        cache->table = rbthash_table_init (GF_SP_CACHE_BUCKETS,
                                                           sp_hashfn, __gf_free,
                                                           0, priv->mem_pool);
                        if (cache->table == NULL) {
                                cache->table = table;
                        } else {
                                rbthash_table_destroy (table);
                                ret = 0;
                                if (priv) {
                                        LOCK (&priv->lock);
                                        {
                                                priv->entries = 0;
                                        }
                                        UNLOCK (&priv->lock);
                                }
                        }
                } else {
                        data = rbthash_remove (cache->table, name,
                                               strlen (name));
                        GF_FREE (data);
                        ret = 0;
                        if (priv) {
                                LOCK (&priv->lock);
                                {
                                        priv->entries--;
                                }
                                UNLOCK (&priv->lock);
                        }
                }
        }
        UNLOCK (&cache->lock);

out:
        return ret;
}


int32_t
sp_cache_get_entry (sp_cache_t *cache, char *name, gf_dirent_t **entry)
{
        int32_t      ret = -1;
        gf_dirent_t *tmp = NULL, *new = NULL;

        GF_VALIDATE_OR_GOTO ("stat-prefetch", cache, out);
        GF_VALIDATE_OR_GOTO ("stat-prefetch", cache->this, out);
        GF_VALIDATE_OR_GOTO (cache->this->name, name, out);
        GF_VALIDATE_OR_GOTO (cache->this->name, entry, out);

        LOCK (&cache->lock);
        {
                tmp = rbthash_get (cache->table, name, strlen (name));
                if (tmp != NULL) {
                        new = gf_dirent_for_name (tmp->d_name);
                        if (new == NULL) {
                                gf_log (cache->this->name, GF_LOG_WARNING,
                                        "cannot create a new dentry to copy "
                                        "from cache");
                                goto unlock;
                        }

                        new->d_ino  = tmp->d_ino;
                        new->d_off  = tmp->d_off;
                        new->d_len  = tmp->d_len;
                        new->d_type = tmp->d_type;
                        new->d_stat = tmp->d_stat;

                        *entry = new;
                        ret = 0;
                }
        }
unlock:
        UNLOCK (&cache->lock);

out:
        return ret;
}


void
sp_cache_free (sp_cache_t *cache)
{
        sp_cache_remove_entry (cache, NULL, 1);
        sp_cache_unref (cache);
}


sp_cache_t *
__sp_get_cache_fd (xlator_t *this, fd_t *fd)
{
        int32_t      ret   = -1;
        sp_cache_t  *cache = NULL;
        uint64_t     value = 0;
        sp_fd_ctx_t *fd_ctx = NULL;

        ret = __fd_ctx_get (fd, this, &value);
        if (ret == -1) {
                goto out;
        }

        fd_ctx = (void *)(long) value;

        cache = fd_ctx->cache;

out:
        return cache;
}


sp_cache_t *
sp_get_cache_fd (xlator_t *this, fd_t *fd)
{
        sp_cache_t  *cache     = NULL;

        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, out);
        GF_VALIDATE_OR_GOTO (this->name, fd, out);

        LOCK (&fd->lock);
        {
                cache = __sp_get_cache_fd (this, fd);
                if (cache != NULL) {
                        sp_cache_ref (cache);
                }
        }
        UNLOCK (&fd->lock);

out:
        return cache;
}


void
sp_fd_ctx_free (sp_fd_ctx_t *fd_ctx)
{
        if (fd_ctx == NULL) {
                goto out;
        }

        if (fd_ctx->parent_inode) {
                inode_unref (fd_ctx->parent_inode);
                fd_ctx->parent_inode = NULL;
        }

        if (fd_ctx->name) {
                GF_FREE (fd_ctx->name);
                fd_ctx->name = NULL;
        }

        if (fd_ctx->cache) {
                sp_cache_free (fd_ctx->cache);
        }

        GF_FREE (fd_ctx);
out:
        return;
}


inline sp_fd_ctx_t *
sp_fd_ctx_init (void)
{
        sp_fd_ctx_t *fd_ctx = NULL;

        fd_ctx = GF_CALLOC (1, sizeof (*fd_ctx), gf_sp_mt_sp_fd_ctx_t);

        return fd_ctx;
}


sp_fd_ctx_t *
sp_fd_ctx_new (xlator_t *this, inode_t *parent, char *name, sp_cache_t *cache)
{
        sp_fd_ctx_t *fd_ctx = NULL;

        fd_ctx = sp_fd_ctx_init ();
        if (fd_ctx == NULL) {
                goto out;
        }

        if (parent) {
                fd_ctx->parent_inode = inode_ref (parent);
        }

        if (name) {
                fd_ctx->name = gf_strdup (name);
                if (fd_ctx->name == NULL) {
                        sp_fd_ctx_free (fd_ctx);
                        fd_ctx = NULL;
                        goto out;
                }
        }

        fd_ctx->cache = cache;

out:
        return fd_ctx;
}


sp_cache_t *
sp_del_cache_fd (xlator_t *this, fd_t *fd)
{
        sp_cache_t  *cache  = NULL;
        uint64_t     value  = 0;
        int32_t      ret    = -1;
        sp_fd_ctx_t *fd_ctx = NULL;

        if (fd == NULL) {
                goto out;
        }

        LOCK (&fd->lock);
        {
                ret = __fd_ctx_get (fd, this, &value);
                if (ret == 0) {
                        fd_ctx = (void *)(long) value;
                        cache = fd_ctx->cache;
                        fd_ctx->cache = NULL;
                }
        }
        UNLOCK (&fd->lock);

out:
        return cache;
}


sp_cache_t *
sp_get_cache_inode (xlator_t *this, inode_t *inode, int32_t pid)
{
        fd_t       *fd    = NULL;
        sp_cache_t *cache = NULL;

        if (inode == NULL) {
                goto out;
        }

        fd = fd_lookup (inode, pid);
        if (fd == NULL) {
                goto out;
        }

        cache = sp_get_cache_fd (this, fd);

        fd_unref (fd);
out:
        return cache;
}


fd_t *
_fd_ref (fd_t *fd);

void
sp_remove_caches_from_all_fds_opened (xlator_t *this, inode_t *inode,
                                      char *name)
{
        fd_t       *fd    = NULL;
        sp_cache_t *cache = NULL;
        struct fd_wrapper {
                fd_t             *fd;
                struct list_head  list;
        };

        struct fd_wrapper *wrapper    = NULL, *tmp = NULL;
        struct list_head   head       = {0, };
        char               remove_all = 0;

        wrapper = NULL;

        INIT_LIST_HEAD (&head);

        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, out);
        GF_VALIDATE_OR_GOTO (this->name, inode, out);

        remove_all = (name == NULL);

        LOCK (&inode->lock);
        {
                list_for_each_entry (fd, &inode->fd_list, inode_list) {
                        wrapper = GF_CALLOC (1, sizeof (*wrapper),
                                             gf_sp_mt_fd_wrapper_t);
                        if (wrapper == NULL) {
                                goto unlock;
                        }

                        INIT_LIST_HEAD (&wrapper->list);

                        wrapper->fd = _fd_ref (fd);
                        list_add_tail (&wrapper->list, &head);
                }
        }
unlock:
        UNLOCK (&inode->lock);

        list_for_each_entry_safe (wrapper, tmp, &head, list) {
                cache = sp_get_cache_fd (this, wrapper->fd);
                if (cache) {
                        sp_cache_remove_entry (cache, name, remove_all);
                        sp_cache_unref (cache);
                }

                list_del (&wrapper->list);
                fd_unref (wrapper->fd);
                GF_FREE (wrapper);
        }

out:
        return;
}


inline int32_t
__sp_put_cache (xlator_t *this, fd_t *fd, sp_cache_t *cache)
{
        sp_fd_ctx_t *fd_ctx = NULL;
        int32_t      ret    = -1;
        uint64_t     value  = 0;

        ret = __fd_ctx_get (fd, this, &value);
        if (!ret) {
                fd_ctx = (void *)(long)value;
        } else {
                fd_ctx = sp_fd_ctx_init ();
                if (fd_ctx == NULL) {
                        ret = -1;
                        goto out;
                }

                ret = __fd_ctx_set (fd, this, (long)(void *)fd_ctx);
                if (ret == -1) {
                        gf_log (this->name, GF_LOG_WARNING,
                                "cannot set stat-prefetch context in fd (%p) "
                                "opened on inode (ino:%"PRId64" gfid:%s)",
                                fd, fd->inode->ino,
                                uuid_utoa (fd->inode->gfid));
                        sp_fd_ctx_free (fd_ctx);
                        goto out;
                }
        }

        if (fd_ctx->cache) {
                sp_cache_free (fd_ctx->cache);
        }

        fd_ctx->cache = cache;

out:
        return ret;
}


inline int32_t
sp_put_cache (xlator_t *this, fd_t *fd, sp_cache_t *cache)
{
        int32_t ret = -1;

        if (fd != NULL) {
                LOCK (&fd->lock);
                {
                        ret = __sp_put_cache (this, fd, cache);
                }
                UNLOCK (&fd->lock);
        }

        return ret;
}


int32_t
sp_cache_add_entries (sp_cache_t *cache, gf_dirent_t *entries)
{
        gf_dirent_t *entry           = NULL, *new = NULL;
        int32_t      ret             = -1;
        uint64_t     expected_offset = 0;
        xlator_t     *this           = NULL;
        sp_private_t *priv           = NULL;

        GF_VALIDATE_OR_GOTO ("stat-prefetch", cache, out);

        this = cache->this;
        if (this && this->private) {
                priv = this->private;
        }

        LOCK (&cache->lock);
        {
                list_for_each_entry (entry, &entries->list, list) {
                        if (IA_ISDIR (entry->d_stat.ia_type)) {
                                continue;
                        }

                        if (uuid_is_null (entry->d_stat.ia_gfid))
                                continue;

                        new = gf_dirent_for_name (entry->d_name);
                        if (new == NULL) {
                                gf_log (cache->this->name, GF_LOG_WARNING,
                                        "cannot create a new dentry to store "
                                        "in cache");
                                goto unlock;
                        }

                        new->d_ino  = entry->d_ino;
                        new->d_off  = entry->d_off;
                        new->d_len  = entry->d_len;
                        new->d_type = entry->d_type;
                        new->d_stat = entry->d_stat;

                        ret = rbthash_insert (cache->table, new, new->d_name,
                                              strlen (new->d_name));
                        if (ret == -1) {
                                gf_log (this->name, GF_LOG_WARNING, "cannot "
                                        "insert dentry (name:%s) into cache",
                                        new->d_name);

                                GF_FREE (new);
                                continue;
                        }

                        expected_offset = new->d_off;
                        if (priv) {
                                LOCK (&priv->lock);
                                {
                                        priv->entries++;
                                }
                                UNLOCK (&priv->lock);
                        }
                }

                cache->expected_offset = expected_offset;

                ret = 0;
        }
unlock:
        UNLOCK (&cache->lock);

out:
        return ret;
}


int32_t
sp_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
               int32_t op_ret, int32_t op_errno, inode_t *inode,
               struct iatt *buf, dict_t *dict, struct iatt *postparent)
{
        struct list_head     waiting_ops = {0, };
        call_stub_t         *stub        = NULL, *tmp = NULL;
        sp_local_t          *local       = NULL;
        int                  need_unwind = 0;
        char                 looked_up   = 0, lookup_in_progress = 0;

        GF_ASSERT (frame);

        INIT_LIST_HEAD (&waiting_ops);

        local = frame->local;
        if (local == NULL) {
                op_ret = -1;
                op_errno = EINVAL;
                gf_log (this->name, GF_LOG_WARNING, "local is NULL, but it is "
                        "needed to find and resume operations waiting on "
                        "lookup");
                goto out;
        }

        if (this == NULL) {
                op_ret = -1;
                op_errno = EINVAL;
                gf_log (frame->this ? frame->this->name : "stat-prefetch",
                        GF_LOG_WARNING, "xlator object (this) is NULL");
                goto out;
        }

        /* For '/' Entry is never cached, don't try to remove it */
        if ((op_ret == -1) && local->loc.parent) {
                sp_remove_caches_from_all_fds_opened (this, local->loc.parent,
                                                      (char *)local->loc.name);
        }

        if (local->is_lookup)
                need_unwind = 1;

        lookup_in_progress = 0;
        looked_up = 1;
        sp_update_inode_ctx (this, local->loc.inode, &op_ret, &op_errno,
                             &lookup_in_progress, &looked_up, buf,
                             &waiting_ops, &op_errno);

        list_for_each_entry_safe (stub, tmp, &waiting_ops, list) {
                list_del_init (&stub->list);
                call_resume (stub);
        }

out:
        if (need_unwind) {
                SP_STACK_UNWIND (lookup, frame, op_ret, op_errno, inode, buf,
                                 dict, postparent);
        }

        return 0;
}


int32_t
sp_get_ancestors (char *path, char **parent, char **grand_parent)
{
        int32_t  ret = -1, i = 0;
        char    *cpy = NULL, *tmp = NULL;

        if (!path || !parent || !grand_parent) {
                ret = 0;
                goto out;
        }

        for (i = 0; i < 2; i++) {
                if (!strcmp (path, "/")) {
                        break;
                }

                tmp = cpy;

                cpy = gf_strdup (path);

                if (tmp != NULL) {
                        GF_FREE (tmp);
                }

                if (cpy == NULL) {
                        ret = -errno;
                        goto out;
                }

                path = dirname (cpy);
                switch (i)
                {
                case 0:
                        *parent = gf_strdup (path);
                        break;
                case 1:
                        *grand_parent = gf_strdup (path);
                        break;
                }
        }

        ret = 0;
out:
        if (cpy != NULL) {
                GF_FREE (cpy);
        }

        return ret;
}


int32_t
sp_cache_remove_parent_entry (call_frame_t *frame, xlator_t *this,
                              inode_table_t *itable, char *path)
{
        char       *parent    = NULL, *grand_parent = NULL, *cpy = NULL;
        inode_t    *inode_gp  = NULL;
        int32_t     ret       = -1;

        ret = sp_get_ancestors (path, &parent, &grand_parent);
        if (ret < 0) {
                goto out;
        }

        if (grand_parent && strcmp (grand_parent, "/")) {
                inode_gp = inode_from_path (itable, grand_parent);
                if (inode_gp) {
                        cpy = gf_strdup (parent);
                        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name,
                                                        cpy, out, ret,
                                                        -ENOMEM);
                        path = basename (cpy);
                        sp_remove_caches_from_all_fds_opened (this, inode_gp,
                                                              path);
                        GF_FREE (cpy);

                        inode_unref (inode_gp);
                }
        }

        ret = 0;
out:
        if (parent) {
                GF_FREE (parent);
        }

        if (grand_parent) {
                GF_FREE (grand_parent);
        }

        return ret;
}


void
sp_is_empty (dict_t *this, char *key, data_t *value, void *data)
{
        char *ptr = data;

        if (strcmp (key, "gfid-req") == 0)
                return;

        if (ptr && *ptr) {
                *ptr = 0;
        }
}


int32_t
sp_lookup_helper (call_frame_t *frame,xlator_t *this, loc_t *loc,
                  dict_t *xattr_req)
{
        uint64_t        value     = 0;
        sp_inode_ctx_t *inode_ctx = NULL;
        int32_t         ret       = 0, op_ret = -1, op_errno = EINVAL;
        call_stub_t    *stub      = NULL;
        char            can_wind  = 0;

        ret = inode_ctx_get (loc->inode, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "stat-prefetch context not set in inode "
                        "(ino:%"PRId64" gfid:%s)", loc->inode->ino,
                        uuid_utoa (loc->inode->gfid));
                op_errno = EINVAL;
                goto unwind;
        }

        inode_ctx = (sp_inode_ctx_t *)(long) value;
        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, inode_ctx, unwind, op_errno,
                                        EINVAL);

        stub = fop_lookup_stub (frame, sp_lookup_helper, loc, xattr_req);
        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, stub, unwind,
                                        op_errno, ENOMEM);

        LOCK (&inode_ctx->lock);
        {
                op_ret = inode_ctx->op_ret;
                op_errno = inode_ctx->op_errno;
                if (op_ret == 0) {
                        if (!inode_ctx->lookup_in_progress) {
                                inode_ctx->lookup_in_progress = 1;
                                can_wind = 1;
                        } else {
                                list_add_tail (&stub->list,
                                               &inode_ctx->waiting_ops);
                                stub = NULL;
                        }
                }
        }
        UNLOCK (&inode_ctx->lock);

        if (op_ret == -1) {
                goto unwind;
        }

        if (can_wind) {
                STACK_WIND (frame, sp_lookup_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->lookup, loc,
                            xattr_req);
        }

        if (stub != NULL) {
                call_stub_destroy (stub);
        }

        return 0;

unwind:
        SP_STACK_UNWIND (lookup, frame, -1, op_errno, NULL, NULL, NULL, NULL);
        if (stub != NULL) {
                call_stub_destroy (stub);
        }

        return 0;
}


/*
 * TODO: implement sending lookups for every fop done on this path. As of now
 * lookup on the path is sent only for the first fop on this path.
 */
int32_t
sp_lookup (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *xattr_req)
{
        gf_dirent_t    *dirent          = NULL;
        char            entry_cached    = 0;
        uint64_t        value           = 0;
        char            xattr_req_empty = 1, can_wind = 0;
        sp_cache_t     *cache           = NULL;
        struct iatt     postparent      = {0, }, buf = {0, };
        int32_t         ret             = -1, op_ret = -1, op_errno = EINVAL;
        sp_inode_ctx_t *inode_ctx       = NULL, *parent_inode_ctx = NULL;
        sp_local_t     *local           = NULL;
        call_stub_t    *stub            = NULL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this->name, this, unwind);
        GF_VALIDATE_OR_GOTO (frame->this->name, loc, unwind);
        GF_VALIDATE_OR_GOTO (frame->this->name, loc->inode, unwind);

        inode_ctx = sp_check_and_create_inode_ctx (this, loc->inode,
                                                   SP_DONT_CARE);
        if (inode_ctx == NULL) {
                op_errno = ENOMEM;
                gf_log (this->name, GF_LOG_WARNING,
                        "cannot create stat-prefetch context in inode (ino:%"
                        PRId64", gfid:%s)(%s)", loc->inode->ino,
                        loc->inode->gfid, strerror (op_errno));
                goto unwind;
        }

        if ((loc->parent == NULL) || (loc->name == NULL)) {
                goto wind;
        }

        if (xattr_req != NULL) {
                dict_foreach (xattr_req, sp_is_empty, &xattr_req_empty);
        }

        if (!xattr_req_empty) {
                goto wind;
        }

        cache = sp_get_cache_inode (this, loc->parent, frame->root->pid);
        if (cache) {
                ret = sp_cache_get_entry (cache, (char *)loc->name, &dirent);
                if (ret == 0) {
                        if (!uuid_is_null (loc->inode->gfid)
                            && (uuid_compare (loc->inode->gfid,
                                              dirent->d_stat.ia_gfid))
                                != 0) {
                                op_ret = -1;
                                op_errno = ESTALE;
                                goto unwind;
                        }

                        ret = inode_ctx_get (loc->parent, this, &value);
                        if ((ret == 0) && (value != 0)) {
                                parent_inode_ctx = (void *)(long)value;
                                postparent = parent_inode_ctx->stbuf;
                                buf = dirent->d_stat;
                                op_ret = 0;
                                op_errno = 0;
                                entry_cached = 1;
                        }

                        GF_FREE (dirent);
                }
        } else if (IA_ISDIR (loc->inode->ia_type)) {
                cache = sp_get_cache_inode (this, loc->inode, frame->root->pid);
                if (cache) {
                        ret = sp_cache_get_entry (cache, ".", &dirent);
                        if (ret == 0) {
                                ret = inode_ctx_get (loc->parent, this, &value);
                                if ((ret == 0) && (value != 0)) {
                                        parent_inode_ctx = (void *)(long)value;
                                        postparent = parent_inode_ctx->stbuf;
                                        buf = dirent->d_stat;
                                        op_ret = 0;
                                        op_errno = 0;
                                        entry_cached = 1;
                                }

                                GF_FREE (dirent);
                        }
                }
        }

wind:
        if (entry_cached) {
                if (cache) {
                        cache->hits++;
                        sp_cache_unref (cache);
                }
        } else {
                if (cache) {
                        cache->miss++;
                        sp_cache_unref (cache);
                }

                stub = fop_lookup_stub (frame, sp_lookup_helper, loc,
                                        xattr_req);
                GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, stub, unwind,
                                                op_errno, ENOMEM);

                local = GF_CALLOC (1, sizeof (*local), gf_sp_mt_sp_local_t);
                GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, local, unwind,
                                                op_errno, ENOMEM);

                frame->local = local;

                ret = loc_copy (&local->loc, loc);
                if (ret == -1) {
                        op_errno = errno;
                        gf_log (this->name, GF_LOG_WARNING,
                                "loc_copy failed (%s)", strerror (op_errno));
                        goto unwind;
                }

                local->is_lookup = 1;

                LOCK (&inode_ctx->lock);
                {
                        if (inode_ctx->lookup_in_progress) {
                                list_add_tail (&stub->list,
                                               &inode_ctx->waiting_ops);
                                stub = NULL;
                        } else {
                                can_wind = 1;
                                inode_ctx->lookup_in_progress = 1;
                        }
                }
                UNLOCK (&inode_ctx->lock);

                if (can_wind) {
                        STACK_WIND (frame, sp_lookup_cbk, FIRST_CHILD(this),
                                    FIRST_CHILD(this)->fops->lookup, loc,
                                    xattr_req);
                }

                if (stub != NULL) {
                        call_stub_destroy (stub);
                }

                return 0;
        }

unwind:
        SP_STACK_UNWIND (lookup, frame, op_ret, op_errno, (loc)?loc->inode:NULL,
                         &buf, NULL, &postparent);

        return 0;
}


int32_t
sp_readdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                int32_t op_ret, int32_t op_errno, gf_dirent_t *entries)
{
        sp_local_t      *local       = NULL;
        sp_cache_t      *cache       = NULL;
        fd_t            *fd          = NULL;
        int32_t          ret         = 0;
        char             was_present = 1;
        sp_private_t    *priv        = NULL;

        GF_ASSERT (frame);
        if (op_ret == -1) {
                goto out;
        }

        if ((this == NULL) || (this->private == NULL)) {
                gf_log (frame->this->name, GF_LOG_WARNING,
                        (this == NULL) ? "xlator object (this) is NULL"
                        : "stat-prefetch configuration (this->private) is "
                        "NULL");
                op_ret = -1;
                op_errno = EINVAL;
                goto out;
        }

        local = frame->local;
        if (local == NULL) {
                gf_log (frame->this->name, GF_LOG_WARNING, "local is NULL");
                op_ret = -1;
                op_errno = EINVAL;
                goto out;
        }

        fd = local->fd;

        priv = this->private;

        LOCK (&priv->lock);
        {
                if (!priv->mem_pool)
                        priv->mem_pool =  mem_pool_new (rbthash_entry_t,
                                                        GF_SP_CACHE_ENTRIES_EXPECTED);
        }
        UNLOCK (&priv->lock);

        if (!priv->mem_pool)
                goto out;

        LOCK (&fd->lock);
        {
                cache = __sp_get_cache_fd (this, fd);
                if (cache == NULL) {
                        was_present = 0;
                        cache = sp_cache_init (this);
                        if (cache == NULL) {
                                gf_log (this->name, GF_LOG_WARNING,
                                        "creation of stat-prefetch cache "
                                        "for fd (%p) opened on inode "
                                        "(ino:%"PRId64", gfid:%s) failed", fd,
                                        fd->inode->ino,
                                        uuid_utoa (fd->inode->gfid));
                                goto unlock;
                        }

                        ret = __sp_put_cache (this, fd, cache);
                        if (ret == -1) {
                                sp_cache_free (cache);
                                gf_log (this->name, GF_LOG_WARNING,
                                        "cannot store cache in fd (%p) opened "
                                        "on inode (ino:%"PRId64", gfid:%s)", fd,
                                        fd->inode->ino,
                                        uuid_utoa (fd->inode->gfid));
                                goto unlock;
                        }
                }

                sp_cache_ref (cache);
        }
unlock:
        UNLOCK (&fd->lock);

        if (cache != NULL) {
                sp_cache_add_entries (cache, entries);
                if (was_present) {
                        sp_cache_unref (cache);
                }
        }

out:
        SP_STACK_UNWIND (readdir, frame, op_ret, op_errno, entries);
        return 0;
}


int32_t
sp_readdir (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
            off_t off)
{
        sp_cache_t *cache    = NULL;
        sp_local_t *local    = NULL;
        char       *path     = NULL;
        int32_t     ret      = -1, op_errno = EINVAL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this->name, this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, fd, unwind);

        cache = sp_get_cache_fd (this, fd);
        if (cache) {
                if (off != cache->expected_offset) {
                        sp_cache_remove_entry (cache, NULL, 1);
                }

                sp_cache_unref (cache);
        }

        ret = inode_path (fd->inode, NULL, &path);
        if (ret < 0) {
                op_errno = -ret;
                gf_log (this->name, GF_LOG_WARNING, "cannot construct path on "
                        "which fd (%p) is opened (fd.inode.ino = %"PRId64", "
                        "fd.inode.gfid = %s) (%s)", fd, fd->inode->ino,
                        uuid_utoa (fd->inode->gfid), strerror (op_errno));
                goto unwind;
        }

        ret = sp_cache_remove_parent_entry (frame, this, fd->inode->table,
                                            path);

        if (ret < 0) {
                op_errno = -ret;
                gf_log (this->name, GF_LOG_WARNING,
                        "cannot remove parent entry from grand-parent's cache"
                        " for path %s", path);
                goto unwind;
        }

        GF_FREE (path);

        local = GF_CALLOC (1, sizeof (*local), gf_sp_mt_sp_local_t);
        if (local) {
                local->fd = fd;
                frame->local = local;
        }

        STACK_WIND (frame, sp_readdir_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->readdirp, fd, size, off);

        return 0;

unwind:
        if (path != NULL) {
                GF_FREE (path);
        }

        SP_STACK_UNWIND (readdir, frame, -1, op_errno, NULL);
        return 0;
}


int32_t
sp_truncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
                 struct iatt *postbuf)
{
        GF_ASSERT (frame);

        SP_STACK_UNWIND (truncate, frame, op_ret, op_errno, prebuf, postbuf);
        return 0;
}


int32_t
sp_rename_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
               int32_t op_ret, int32_t op_errno, struct iatt *buf,
               struct iatt *preoldparent, struct iatt *postoldparent,
               struct iatt *prenewparent, struct iatt *postnewparent)
{
        GF_ASSERT (frame);
        SP_STACK_UNWIND (rename, frame, op_ret, op_errno, buf, preoldparent,
                         postoldparent, prenewparent, postnewparent);
        return 0;
}


int32_t
sp_fd_cbk (call_frame_t *frame, void *cookie, xlator_t *this, int32_t op_ret,
           int32_t op_errno, fd_t *fd)
{
        sp_local_t  *local = NULL;
        sp_fd_ctx_t *fd_ctx = NULL;

        GF_ASSERT (frame);

        if (op_ret == -1) {
                goto out;
        }

        if (this == NULL) {
                gf_log (frame->this ? frame->this->name : "stat-prefetch",
                        GF_LOG_WARNING, "xlator object (this) is NULL");
                op_ret = -1;
                op_errno = EINVAL;
                goto out;
        }

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                op_ret = -1;
                op_errno = EINVAL;
                goto out;
        }

        if (fd == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "fd is NULL");
                op_ret = -1;
                op_errno = EINVAL;
                goto out;
        }

        fd_ctx = sp_fd_ctx_new (this, local->loc.parent,
                                (char *)local->loc.name, NULL);
        if (fd_ctx == NULL) {
                op_ret = -1;
                op_errno = ENOMEM;
                goto out;
        }

        op_ret = fd_ctx_set (fd, this, (long)(void *)fd_ctx);
        if (op_ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "cannot set stat-prefetch context in fd (%p) opened on "
                        "inode (ino:%"PRId64", gfid:%s)", fd, fd->inode->ino,
                        uuid_utoa (fd->inode->gfid));
                sp_fd_ctx_free (fd_ctx);
                op_errno = ENOMEM;
        }

out:
        SP_STACK_UNWIND (open, frame, op_ret, op_errno, fd);
        return 0;
}


int32_t
sp_open_helper (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
                fd_t *fd, int32_t wbflags)
{
        uint64_t        value     = 0;
        sp_inode_ctx_t *inode_ctx = NULL;
        int32_t         ret       = 0, op_ret = -1, op_errno = EINVAL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this->name, this, unwind);
        GF_VALIDATE_OR_GOTO (frame->this->name, loc, unwind);

        ret = inode_ctx_get (loc->inode, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "stat-prefetch context not set in inode "
                        "(ino:%"PRId64" gfid:%s)", loc->inode->ino,
                        uuid_utoa (loc->inode->gfid));
                op_errno = EINVAL;
                goto unwind;
        }

        inode_ctx = (sp_inode_ctx_t *)(long) value;
        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, inode_ctx, unwind, op_errno,
                                        EINVAL);

        LOCK (&inode_ctx->lock);
        {
                op_ret = inode_ctx->op_ret;
                op_errno = inode_ctx->op_errno;
        }
        UNLOCK (&inode_ctx->lock);

        if ((op_ret == -1) && ((op_errno != ENOENT)
                               || !((op_errno == ENOENT)
                                    && (flags & O_CREAT)))) {
                gf_log (this->name, GF_LOG_WARNING, "lookup-behind has failed "
                        "for path (%s)(%s), unwinding open call waiting on "
                        "it", loc->path, strerror (op_errno));
                goto unwind;
        }

        STACK_WIND (frame, sp_fd_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->open, loc, flags, fd, wbflags);

        return 0;

unwind:
        SP_STACK_UNWIND (open, frame, -1, op_errno, fd);
        return 0;
}


int32_t
sp_open (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
         fd_t *fd, int wbflags)
{
        call_stub_t    *stub         = NULL;
        sp_local_t     *local        = NULL;
        int32_t         op_errno     = EINVAL, ret = -1;
        char            can_wind     = 0, need_lookup = 0, need_unwind = 1;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this->name, this, out);
        GF_VALIDATE_OR_GOTO (this->name, loc, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->inode, out);

        local = GF_CALLOC (1, sizeof (*local), gf_sp_mt_sp_local_t);
        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, local, out, op_errno,
                                        ENOMEM);

        frame->local = local;

        ret = loc_copy (&local->loc, loc);
        if (ret == -1) {
                op_errno = errno;
                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed (%s)",
                        strerror (op_errno));
                goto out;
        }

        stub = fop_open_stub (frame, sp_open_helper, loc, flags, fd, wbflags);
        if (stub == NULL) {
                op_errno = ENOMEM;
                goto out;
        }

        sp_process_inode_ctx (frame, this, loc, stub, &need_unwind,
                              &need_lookup, &can_wind, &op_errno);
out:
        if (need_unwind) {
                SP_STACK_UNWIND (open, frame, -1, op_errno, fd);
        } else if (need_lookup) {
                STACK_WIND (frame, sp_lookup_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->lookup, loc, NULL);
        } else if (can_wind) {
                STACK_WIND (frame, sp_fd_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->open, loc, flags, fd,
                            wbflags);
        }

        return 0;

}

static int32_t
sp_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
               int32_t op_ret, int32_t op_errno, fd_t *fd, inode_t *inode,
               struct iatt *buf, struct iatt *preparent,
               struct iatt *postparent)
{
        sp_local_t     *local              = NULL;
        sp_fd_ctx_t    *fd_ctx             = NULL;
        char            lookup_in_progress = 0, looked_up = 0;

        GF_ASSERT (frame);

        if (op_ret == -1) {
                goto out;
        }

        if (this == NULL) {
                gf_log (frame->this ? frame->this->name : "stat-prefetch",
                        GF_LOG_WARNING, "xlator object (this) is NULL");
                op_ret = -1;
                op_errno = EINVAL;
                goto out;
        }

        local = frame->local;
        if (local == NULL) {
                gf_log (this->name, GF_LOG_WARNING, "local is NULL");
                op_ret = -1;
                op_errno = EINVAL;
                goto out;
        }

        looked_up = 1;
        op_ret = sp_update_inode_ctx (this, local->loc.inode, &op_ret,
                                      &op_errno, &lookup_in_progress,
                                      &looked_up, buf, NULL, &op_errno);
        if (op_ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "updating stat-prefetch context in inode (ino:%"
                        PRId64", gfid:%s) (path: %s) failed (%s)",
                        local->loc.inode->ino,
                        uuid_utoa (local->loc.inode->gfid), local->loc.path,
                        strerror (op_errno));
                goto out;
        }

        op_ret = sp_update_inode_ctx (this, local->loc.parent, NULL, NULL, NULL,
                                      NULL, postparent, NULL, &op_errno);
        if (op_ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "updating stat-prefetch context in parent inode failed "
                        "for path (%s)(%s)", local->loc.path,
                        strerror (op_errno));
                goto out;
        }

        fd_ctx = sp_fd_ctx_new (this, local->loc.parent,
                                (char *)local->loc.name, NULL);
        if (fd_ctx == NULL) {
                op_ret = -1;
                op_errno = ENOMEM;
                goto out;
        }

        op_ret = fd_ctx_set (fd, this, (long)(void *)fd_ctx);
        if (op_ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "cannot set stat-prefetch context in fd (%p) opened on "
                        "inode (ino:%"PRId64", gfid:%s)", fd, fd->inode->ino,
                        uuid_utoa (fd->inode->gfid));
                sp_fd_ctx_free (fd_ctx);
                op_errno = ENOMEM;
        }

out:
        SP_STACK_UNWIND (create, frame, op_ret, op_errno, fd, inode, buf,
                         preparent, postparent);
        return 0;
}


int32_t
sp_create (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
           mode_t mode, fd_t *fd, dict_t *params)
{
        sp_local_t        *local        = NULL;
        int32_t            op_errno     = -1, ret = -1;
        char               need_unwind  = 1;
        sp_inode_ctx_t    *inode_ctx    = NULL;

        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, loc, out, op_errno,
                                        EINVAL);
        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, loc->path, out, op_errno,
                                        EINVAL);
        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, loc->name, out, op_errno,
                                        EINVAL);
        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, loc->inode, out,
                                        op_errno, EINVAL);

        ret = sp_cache_remove_parent_entry (frame, this, loc->inode->table,
                                            (char *)loc->path);
        if (ret == -1) {
                op_errno = ENOMEM;
                goto out;
        }

        local = GF_CALLOC (1, sizeof (*local), gf_sp_mt_sp_local_t);
        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, local, out, op_errno,
                                        ENOMEM);

        frame->local = local;

        ret = loc_copy (&local->loc, loc);
        if (ret == -1) {
                op_errno = errno;
                gf_log (this->name, GF_LOG_WARNING,
                        "loc_copy failed (%s)", strerror (op_errno));
                goto out;
        }

        inode_ctx = sp_check_and_create_inode_ctx (this, loc->inode,
                                                   SP_DONT_EXPECT);
        if (inode_ctx == NULL) {
                op_errno = ENOMEM;
                gf_log (this->name, GF_LOG_WARNING,
                        "cannot create stat-prefetch context in inode (ino:%"
                        PRId64", gfid:%s)(%s)", loc->inode->ino,
                        loc->inode->gfid, strerror (op_errno));
                goto out;
        }

        need_unwind = 0;
out:
        if (need_unwind) {
                SP_STACK_UNWIND (create, frame, -1, op_errno, NULL, NULL, NULL,
                                 NULL, NULL);
        } else {
                STACK_WIND (frame, sp_create_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->create, loc, flags,
                            mode, fd, params);
        }
        return 0;
}


int32_t
sp_opendir_helper (call_frame_t *frame, xlator_t *this, loc_t *loc, fd_t *fd)
{
        uint64_t        value     = 0;
        sp_inode_ctx_t *inode_ctx = NULL;
        int32_t         ret       = 0, op_ret = -1, op_errno = EINVAL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this ? frame->this->name : "stat-prefetch",
                             this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, loc, unwind);

        ret = inode_ctx_get (loc->inode, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "stat-prefetch context not set in inode "
                        "(ino:%"PRId64" gfid:%s)", loc->inode->ino,
                        uuid_utoa (loc->inode->gfid));
                goto unwind;
        }

        inode_ctx = (sp_inode_ctx_t *)(long) value;
        GF_VALIDATE_OR_GOTO (this->name, inode_ctx, unwind);

        LOCK (&inode_ctx->lock);
        {
                op_ret = inode_ctx->op_ret;
                op_errno = inode_ctx->op_errno;
        }
        UNLOCK (&inode_ctx->lock);

        if (op_ret == -1) {
                gf_log (this->name, GF_LOG_WARNING, "lookup-behind has failed "
                        "for path (%s)(%s), unwinding opendir call waiting "
                        "on it", loc->path, strerror (op_errno));
                goto unwind;
        }

        STACK_WIND (frame, sp_fd_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->opendir, loc, fd);

        return 0;

unwind:
        SP_STACK_UNWIND (opendir, frame, -1, op_errno, NULL);
        return 0;
}


int32_t
sp_opendir (call_frame_t *frame, xlator_t *this, loc_t *loc, fd_t *fd)
{
        sp_local_t     *local        = NULL;
        call_stub_t    *stub         = NULL;
        int32_t         op_errno     = EINVAL, ret = -1;
        char            can_wind     = 0, need_lookup = 0, need_unwind = 1;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this ? frame->this->name : "stat-prefetch",
                             this, out);
        GF_VALIDATE_OR_GOTO (this->name, loc, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->inode, out);

        local = GF_CALLOC (1, sizeof (*local), gf_sp_mt_sp_local_t);
        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, local, out, op_errno,
                                        ENOMEM);

        frame->local = local;

        ret = loc_copy (&local->loc, loc);
        if (ret == -1) {
                op_errno = errno;
                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed (%s)",
                        strerror (op_errno));
                goto out;
        }

        stub = fop_opendir_stub (frame, sp_opendir_helper, loc, fd);
        if (stub == NULL) {
                op_errno = ENOMEM;
                goto out;
        }

        sp_process_inode_ctx (frame, this, loc, stub, &need_unwind,
                              &need_lookup, &can_wind, &op_errno);

out:
        if (need_unwind) {
                SP_STACK_UNWIND (opendir, frame, -1, op_errno, NULL);
        } else if (need_lookup) {
                STACK_WIND (frame, sp_lookup_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->lookup, loc, NULL);
        } else if (can_wind) {
                STACK_WIND (frame, sp_fd_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->opendir, loc, fd);
        }

        return 0;
}


int32_t
sp_new_entry_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                  int32_t op_ret, int32_t op_errno, inode_t *inode,
                  struct iatt *buf, struct iatt *preparent,
                  struct iatt *postparent)
{
        sp_local_t *local              = NULL;
        char        lookup_in_progress = 0, looked_up = 0;

        GF_ASSERT (frame);

        if (op_ret == -1) {
                goto out;
        }

        local = frame->local;
        if (local == NULL) {
                gf_log (frame->this->name, GF_LOG_WARNING, "local is NULL");
                op_ret = -1;
                op_errno = EINVAL;
                goto out;
        }

        if (this == NULL) {
                gf_log (frame->this->name, GF_LOG_WARNING,
                        "xlator object (this) is NULL");
                op_ret = -1;
                op_errno = EINVAL;
                goto out;
        }

        looked_up = 1;
        op_ret = sp_update_inode_ctx (this, local->loc.inode, &op_ret,
                                      &op_errno, &lookup_in_progress,
                                      &looked_up, buf, NULL, &op_errno);
        if (op_ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "updating stat-prefetch context in inode (ino:%"
                        PRId64", gfid:%s) (path: %s) failed (%s)",
                        local->loc.inode->ino,
                        uuid_utoa (local->loc.inode->gfid), local->loc.path,
                        strerror (op_errno));
                goto out;
        }

        op_ret = sp_update_inode_ctx (this, local->loc.parent, NULL, NULL, NULL,
                                      NULL, postparent, NULL, &op_errno);
        if (op_ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "updating stat-prefetch context in parent inode failed "
                        "for path (%s)(%s)", local->loc.path,
                        strerror (op_errno));
        }

out:
        SP_STACK_UNWIND (mkdir, frame, op_ret, op_errno, inode, buf, preparent,
                         postparent);
        return 0;
}


int
sp_mkdir (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
          dict_t *params)
{
        int32_t         ret          = -1, op_errno = EINVAL;
        char            need_unwind  = 1;
        sp_inode_ctx_t *inode_ctx    = NULL;
        sp_local_t     *local        = NULL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this->name, this, out);
        GF_VALIDATE_OR_GOTO (this->name, loc, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->path, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->name, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->inode, out);

        ret = sp_cache_remove_parent_entry (frame, this, loc->inode->table,
                                            (char *)loc->path);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "cannot remove parent entry from grand-parent's cache "
                        "for path (%s)", loc->path);
                op_errno = ENOMEM;
                goto out;
        }

        local = GF_CALLOC (1, sizeof (*local), gf_sp_mt_sp_local_t);
        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, local, out, op_errno,
                                        ENOMEM);

        frame->local = local;

        ret = loc_copy (&local->loc, loc);
        if (ret == -1) {
                op_errno = errno;
                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed (%s)",
                        strerror (op_errno));
                goto out;
        }

        inode_ctx = sp_check_and_create_inode_ctx (this, loc->inode,
                                                   SP_DONT_EXPECT);
        if (inode_ctx == NULL) {
                op_errno = ENOMEM;
                gf_log (this->name, GF_LOG_WARNING,
                        "cannot create stat-prefetch context in inode (ino:%"
                        PRId64", gfid:%s)(%s)", loc->inode->ino,
                        loc->inode->gfid, strerror (op_errno));
                goto out;
        }

        need_unwind = 0;
out:
        if (need_unwind) {
                SP_STACK_UNWIND (mkdir, frame, -1, op_errno, NULL, NULL, NULL,
                                 NULL);
        } else {
                STACK_WIND (frame, sp_new_entry_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->mkdir, loc, mode, params);
        }

        return 0;
}


int
sp_mknod (call_frame_t *frame, xlator_t *this, loc_t *loc, mode_t mode,
          dev_t rdev, dict_t *params)
{
        int32_t         op_errno     = EINVAL, ret = -1;
        char            need_unwind  = 1;
        sp_inode_ctx_t *inode_ctx    = NULL;
        sp_local_t     *local        = NULL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this ? frame->this->name : "stat-prefetch",
                             this, out);
        GF_VALIDATE_OR_GOTO (this->name, loc, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->path, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->name, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->inode, out);

        ret = sp_cache_remove_parent_entry (frame, this, loc->inode->table,
                                            (char *)loc->path);
        if (ret == -1) {
                op_errno = ENOMEM;
                gf_log (this->name, GF_LOG_WARNING,
                        "cannot remove parent entry from grand-parent's cache "
                        "for path (%s)", loc->path);
                goto out;
        }

        local = GF_CALLOC (1, sizeof (*local), gf_sp_mt_sp_local_t);
        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, local, out, op_errno,
                                        ENOMEM);

        frame->local = local;

        ret = loc_copy (&local->loc, loc);
        if (ret == -1) {
                op_errno = errno;
                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed (%s)",
                        strerror (op_errno));
                goto out;
        }

        inode_ctx = sp_check_and_create_inode_ctx (this, loc->inode,
                                                   SP_DONT_EXPECT);
        if (inode_ctx == NULL) {
                op_errno = ENOMEM;
                gf_log (this->name, GF_LOG_WARNING,
                        "cannot create stat-prefetch context in inode (ino:%"
                        PRId64", gfid:%s)(%s)", loc->inode->ino,
                        loc->inode->gfid, strerror (op_errno));
                goto out;
        }

        need_unwind = 0;
out:
        if (need_unwind) {
                SP_STACK_UNWIND (mknod, frame, -1, op_errno, NULL, NULL, NULL,
                                 NULL);
        } else {
                STACK_WIND (frame, sp_new_entry_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->mknod, loc, mode,
                            rdev, params);
        }

        return 0;
}


int
sp_symlink (call_frame_t *frame, xlator_t *this, const char *linkpath,
            loc_t *loc, dict_t *params)
{
        int32_t         ret          = -1, op_errno = EINVAL;
        char            need_unwind  = 1;
        sp_inode_ctx_t *inode_ctx    = NULL;
        sp_local_t     *local        = NULL;

        GF_ASSERT (frame);

        GF_VALIDATE_OR_GOTO ((frame->this ? frame->this->name
                              : "stat-prefetch"),
                             this, out);
        GF_VALIDATE_OR_GOTO (this->name, loc, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->path, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->name, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->inode, out);

        ret = sp_cache_remove_parent_entry (frame, this, loc->inode->table,
                                            (char *)loc->path);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "cannot remove parent entry from grand-parent's cache "
                        "for path (%s)", loc->path);
                op_errno = ENOMEM;
                goto out;
        }

        local = GF_CALLOC (1, sizeof (*local), gf_sp_mt_sp_local_t);
        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, local, out, op_errno,
                                        ENOMEM);

        frame->local = local;

        ret = loc_copy (&local->loc, loc);
        if (ret == -1) {
                op_errno = errno;
                gf_log (this->name, GF_LOG_WARNING, "loc_copy failed (%s)",
                        strerror (op_errno));
                goto out;
        }

        inode_ctx = sp_check_and_create_inode_ctx (this, loc->inode,
                                                   SP_DONT_EXPECT);
        if (inode_ctx == NULL) {
                op_errno = ENOMEM;
                gf_log (this->name, GF_LOG_WARNING,
                        "cannot create stat-prefetch context in inode (ino:%"
                        PRId64", gfid:%s)(%s)", loc->inode->ino,
                        loc->inode->gfid, strerror (op_errno));
                goto out;
        }

        need_unwind = 0;
out:
        if (need_unwind) {
                SP_STACK_UNWIND (symlink, frame, -1, op_errno, NULL, NULL, NULL,
                                 NULL);
        } else {
                STACK_WIND (frame, sp_new_entry_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->symlink, linkpath, loc,
                            params);
        }

        return 0;
}


int32_t
sp_link_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
             int32_t op_ret, int32_t op_errno, inode_t *inode,
             struct iatt *buf, struct iatt *preparent,
             struct iatt *postparent)
{
        GF_ASSERT (frame);
        SP_STACK_UNWIND (link, frame, op_ret, op_errno, inode, buf, preparent,
                         postparent);
        return 0;
}


int32_t
sp_link_helper (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
                loc_t *newloc)
{
        uint64_t        value     = 0;
        sp_inode_ctx_t *inode_ctx = NULL;
        int32_t         ret       = 0, op_ret = -1, op_errno = EINVAL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this ? frame->this->name : "stat-prefetch",
                             this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, oldloc, unwind);
        GF_VALIDATE_OR_GOTO (this->name, newloc, unwind);

        ret = inode_ctx_get (oldloc->inode, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "stat-prefetch context not set in inode "
                        "(ino:%"PRId64" gfid:%s)", oldloc->inode->ino,
                        uuid_utoa (oldloc->inode->gfid));
                goto unwind;
        }

        inode_ctx = (sp_inode_ctx_t *)(long) value;
        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, inode_ctx, unwind, op_errno,
                                        EINVAL);

        LOCK (&inode_ctx->lock);
        {
                op_ret = inode_ctx->op_ret;
                op_errno = inode_ctx->op_errno;
        }
        UNLOCK (&inode_ctx->lock);

        if (op_ret == -1) {
                gf_log (this->name, GF_LOG_WARNING, "lookup-behind has failed "
                        "for path (%s)(%s), unwinding link call waiting on "
                        "it", oldloc->path, strerror (op_errno));
                goto unwind;
        }

        STACK_WIND (frame, sp_link_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->link, oldloc, newloc);

        return 0;

unwind:
        SP_STACK_UNWIND (link, frame, -1, op_errno, NULL, NULL, NULL, NULL);
        return 0;
}


int32_t
sp_link (call_frame_t *frame, xlator_t *this, loc_t *oldloc, loc_t *newloc)
{
        call_stub_t *stub         = NULL;
        int32_t      ret          = 0, op_errno = EINVAL;
        char         can_wind     = 0, need_lookup = 0, need_unwind = 1;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this ? frame->this->name : "stat-prefetch",
                             this, out);
        GF_VALIDATE_OR_GOTO (this->name, newloc, out);
        GF_VALIDATE_OR_GOTO (this->name, newloc->path, out);
        GF_VALIDATE_OR_GOTO (this->name, newloc->name, out);
        GF_VALIDATE_OR_GOTO (this->name, newloc->inode, out);
        GF_VALIDATE_OR_GOTO (this->name, oldloc->name, out);

        ret = sp_cache_remove_parent_entry (frame, this, newloc->parent->table,
                                            (char *)newloc->path);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "cannot remove parent entry from grand-parent's cache "
                        "for path (%s)", newloc->path);
                op_errno = ENOMEM;
                goto out;
        }

        sp_remove_caches_from_all_fds_opened (this, oldloc->parent,
                                              (char *)oldloc->name);

        stub = fop_link_stub (frame, sp_link_helper, oldloc, newloc);
        if (stub == NULL) {
                op_errno = ENOMEM;
                goto out;
        }

        sp_process_inode_ctx (frame, this, oldloc, stub, &need_unwind,
                              &need_lookup, &can_wind, &op_errno);

out:
        if (need_unwind) {
                SP_STACK_UNWIND (link, frame, -1, op_errno, NULL, NULL, NULL,
                                 NULL);
        } else if (need_lookup) {
                STACK_WIND (frame, sp_lookup_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->lookup, oldloc, NULL);
        } else if (can_wind) {
                STACK_WIND (frame, sp_link_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->link, oldloc, newloc);
        }

        return 0;
}


int32_t
sp_truncate_helper (call_frame_t *frame, xlator_t *this, loc_t *loc,
                    off_t offset)
{
        uint64_t        value     = 0;
        sp_inode_ctx_t *inode_ctx = NULL;
        int32_t         ret       = 0, op_ret = -1, op_errno = EINVAL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this ? frame->this->name : "stat-prefetch",
                             this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, loc, unwind);

        ret = inode_ctx_get (loc->inode, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "stat-prefetch context not set in inode "
                        "(ino:%"PRId64" gfid:%s)", loc->inode->ino,
                        uuid_utoa (loc->inode->gfid));
                goto unwind;
        }

        inode_ctx = (sp_inode_ctx_t *)(long) value;
        GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, inode_ctx, unwind, op_errno,
                                        EINVAL);

        LOCK (&inode_ctx->lock);
        {
                op_ret = inode_ctx->op_ret;
                op_errno = inode_ctx->op_errno;
        }
        UNLOCK (&inode_ctx->lock);

        if (op_ret == -1) {
                gf_log (this->name, GF_LOG_WARNING, "lookup-behind has failed "
                        "for path (%s)(%s), unwinding truncate call "
                        "waiting on it", loc->path, strerror (op_errno));
                goto unwind;
        }

        STACK_WIND (frame, sp_truncate_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->truncate, loc, offset);

        return 0;

unwind:
        SP_STACK_UNWIND (truncate, frame, -1, op_errno, NULL, NULL);
        return 0;
}


int32_t
sp_truncate (call_frame_t *frame, xlator_t *this, loc_t *loc, off_t offset)
{
        int32_t         op_errno     = EINVAL;
        call_stub_t    *stub         = NULL;
        char            can_wind     = 0, need_lookup = 0, need_unwind = 1;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this ? frame->this->name : "stat-prefetch",
                             this, out);
        GF_VALIDATE_OR_GOTO (this->name, loc, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->name, out);

        sp_remove_caches_from_all_fds_opened (this, loc->parent,
                                              (char *)loc->name);

        stub = fop_truncate_stub (frame, sp_truncate_helper, loc, offset);
        if (stub == NULL) {
                op_errno = ENOMEM;
                goto out;
        }

        sp_process_inode_ctx (frame, this, loc, stub, &need_unwind,
                              &need_lookup, &can_wind, &op_errno);

out:
        if (need_unwind) {
                SP_STACK_UNWIND (truncate, frame, -1, op_errno, NULL, NULL);
        } else if (need_lookup) {
                STACK_WIND (frame, sp_lookup_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->lookup, loc, NULL);
        } else if (can_wind) {
                STACK_WIND (frame, sp_truncate_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->truncate, loc, offset);
        }

        return 0;
}


int32_t
sp_ftruncate (call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset)
{
        sp_fd_ctx_t *fd_ctx = NULL;
        uint64_t     value  = 0;
        int32_t      ret    = 0, op_errno = EINVAL;
        inode_t     *parent = NULL;
        char        *name   = NULL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this ? frame->this->name : "stat-prefetch",
                             this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, fd, unwind);

        ret = fd_ctx_get (fd, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING, "stat-prefetch context not "
                        "set in fd (%p) opened on inode (ino:%"PRId64", "
                        "gfid:%s", fd, fd->inode->ino,
                        uuid_utoa (fd->inode->gfid));
                goto unwind;
        }

        fd_ctx = (void *)(long)value;
        name   = fd_ctx->name;
        parent = fd_ctx->parent_inode;

        sp_remove_caches_from_all_fds_opened (this, parent, (char *)name);

        STACK_WIND (frame, sp_truncate_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->ftruncate, fd, offset);
        return 0;

unwind:
        SP_STACK_UNWIND (ftruncate, frame, -1, op_errno, NULL, NULL);
        return 0;
}


int32_t
sp_setattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                int32_t op_ret, int32_t op_errno,
                struct iatt *prestat, struct iatt *poststat)
{
        GF_ASSERT (frame);
        SP_STACK_UNWIND (setattr, frame, op_ret, op_errno, prestat, poststat);
        return 0;
}


int
sp_setattr_helper (call_frame_t *frame, xlator_t *this,
                   loc_t *loc, struct iatt *buf, int32_t valid)
{
        uint64_t        value     = 0;
        sp_inode_ctx_t *inode_ctx = NULL;
        int32_t         ret       = 0, op_ret = -1, op_errno = EINVAL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this ? frame->this->name : "stat-prefetch",
                             this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, loc, unwind);

        ret = inode_ctx_get (loc->inode, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "stat-prefetch context not set in inode "
                        "(ino:%"PRId64" gfid:%s)", loc->inode->ino,
                        uuid_utoa (loc->inode->gfid));
                op_errno = EINVAL;
                goto unwind;
        }

        inode_ctx = (sp_inode_ctx_t *)(long) value;
        GF_VALIDATE_OR_GOTO (this->name, inode_ctx, unwind);

        LOCK (&inode_ctx->lock);
        {
                op_ret = inode_ctx->op_ret;
                op_errno = inode_ctx->op_errno;
        }
        UNLOCK (&inode_ctx->lock);

        if (op_ret == -1) {
                gf_log (this->name, GF_LOG_WARNING, "lookup-behind has failed "
                        "for path (%s)(%s), unwinding setattr call "
                        "waiting on it", loc->path, strerror (op_errno));
                goto unwind;
        }

        STACK_WIND (frame, sp_setattr_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->setattr, loc, buf, valid);

        return 0;

unwind:
        SP_STACK_UNWIND (setattr, frame, -1, op_errno, NULL, NULL);
        return 0;
}


int
sp_setattr (call_frame_t *frame, xlator_t *this,
            loc_t *loc, struct iatt *buf, int32_t valid)
{
        int32_t         op_errno     = EINVAL;
        call_stub_t    *stub         = NULL;
        char            can_wind     = 0, need_lookup = 0, need_unwind = 1;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this ? frame->this->name : "stat-prefetch",
                             this, out);
        GF_VALIDATE_OR_GOTO (this->name, loc, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->name, out);

        sp_remove_caches_from_all_fds_opened (this, loc->parent,
                                              (char *)loc->name);

        stub = fop_setattr_stub (frame, sp_setattr_helper, loc, buf, valid);
        if (stub == NULL) {
                op_errno = ENOMEM;
                goto out;
        }

        sp_process_inode_ctx (frame, this, loc, stub, &need_unwind,
                              &need_lookup, &can_wind, &op_errno);

out:
        if (need_unwind) {
                SP_STACK_UNWIND (setattr, frame, -1, op_errno, NULL, NULL);
        } else if (need_lookup) {
                STACK_WIND (frame, sp_lookup_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->lookup, loc, NULL);
        } else if (can_wind) {
                STACK_WIND (frame, sp_setattr_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->setattr, loc, buf, valid);
        }

        return 0;
}


int32_t
sp_readlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 int32_t op_ret, int32_t op_errno, const char *path,
                 struct iatt *buf)
{
        GF_ASSERT (frame);
        SP_STACK_UNWIND (readlink, frame, op_ret, op_errno, path, buf);
        return 0;
}


int32_t
sp_readlink_helper (call_frame_t *frame, xlator_t *this, loc_t *loc,
                    size_t size)
{
        uint64_t        value     = 0;
        sp_inode_ctx_t *inode_ctx = NULL;
        int32_t         ret       = 0, op_ret = -1, op_errno = EINVAL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this ? frame->this->name : "stat-prefetch",
                             this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, loc, unwind);

        ret = inode_ctx_get (loc->inode, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "stat-prefetch context not set in inode "
                        "(ino:%"PRId64" gfid:%s)", loc->inode->ino,
                        uuid_utoa (loc->inode->gfid));
                goto unwind;
        }

        inode_ctx = (sp_inode_ctx_t *)(long) value;
        GF_VALIDATE_OR_GOTO (this->name, inode_ctx, unwind);

        LOCK (&inode_ctx->lock);
        {
                op_ret = inode_ctx->op_ret;
                op_errno = inode_ctx->op_errno;
        }
        UNLOCK (&inode_ctx->lock);

        if (op_ret == -1) {
                gf_log (this->name, GF_LOG_WARNING, "lookup-behind has failed "
                        "for path (%s)(%s), unwinding readlink call "
                        "waiting on it", loc->path, strerror (op_errno));
                goto unwind;
        }

        STACK_WIND (frame, sp_readlink_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->readlink, loc, size);

        return 0;

unwind:
        SP_STACK_UNWIND (readlink, frame, -1, op_errno, NULL, NULL);
        return 0;
}


int32_t
sp_readlink (call_frame_t *frame, xlator_t *this, loc_t *loc, size_t size)
{
        int32_t         op_errno     = EINVAL;
        call_stub_t    *stub         = NULL;
        char            can_wind     = 0, need_lookup = 0, need_unwind = 1;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, out);
        GF_VALIDATE_OR_GOTO (this->name, loc, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->name, out);

        sp_remove_caches_from_all_fds_opened (this, loc->parent,
                                              (char *)loc->name);

        stub = fop_readlink_stub (frame, sp_readlink_helper, loc, size);
        if (stub == NULL) {
                op_errno = ENOMEM;
                goto out;
        }

        sp_process_inode_ctx (frame, this, loc, stub, &need_unwind,
                              &need_lookup, &can_wind, &op_errno);

out:
        if (need_unwind) {
                SP_STACK_UNWIND (readlink, frame, -1, op_errno, NULL, NULL);
        } else if (need_lookup) {
                STACK_WIND (frame, sp_lookup_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->lookup, loc, NULL);
        } else if (can_wind) {
                STACK_WIND (frame, sp_readlink_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->readlink, loc, size);
        }

        return 0;
}


int32_t
sp_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
               int32_t op_ret, int32_t op_errno, struct iatt *preparent,
               struct iatt *postparent)
{
        GF_ASSERT (frame);
        SP_STACK_UNWIND (unlink, frame, op_ret, op_errno, preparent,
                         postparent);
        return 0;
}


int32_t
sp_err_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
            int32_t op_ret, int32_t op_errno)
{
        GF_ASSERT (frame);
        SP_STACK_UNWIND (setxattr, frame, op_ret, op_errno);
        return 0;
}


int32_t
sp_unlink_helper (call_frame_t *frame, xlator_t *this, loc_t *loc)
{
        uint64_t        value     = 0;
        sp_inode_ctx_t *inode_ctx = NULL;
        int32_t         ret       = 0, op_ret = -1, op_errno = EINVAL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, loc, unwind);

        ret = inode_ctx_get (loc->inode, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "stat-prefetch context not set in inode "
                        "(ino:%"PRId64" gfid:%s)", loc->inode->ino,
                        uuid_utoa (loc->inode->gfid));
                goto unwind;
        }

        inode_ctx = (sp_inode_ctx_t *)(long) value;
        GF_VALIDATE_OR_GOTO (this->name, inode_ctx, unwind);

        LOCK (&inode_ctx->lock);
        {
                op_ret = inode_ctx->op_ret;
                op_errno = inode_ctx->op_errno;
        }
        UNLOCK (&inode_ctx->lock);

        if (op_ret == -1) {
                gf_log (this->name, GF_LOG_WARNING, "lookup-behind has failed "
                        "for path (%s)(%s), unwinding unlink call "
                        "waiting on it", loc->path, strerror (op_errno));
                goto unwind;
        }

        STACK_WIND (frame, sp_unlink_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->unlink, loc);

        return 0;

unwind:
        SP_STACK_UNWIND (unlink, frame, -1, op_errno, NULL, NULL);
        return 0;
}


int32_t
sp_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc)
{
        int32_t         ret          = -1, op_errno = EINVAL;
        call_stub_t    *stub         = NULL;
        char            can_wind     = 0, need_lookup = 0, need_unwind = 1;

        GF_VALIDATE_OR_GOTO (this->name, loc, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->name, out);

        sp_remove_caches_from_all_fds_opened (this, loc->parent,
                                              (char *)loc->name);

        ret = sp_cache_remove_parent_entry (frame, this, loc->parent->table,
                                            (char *)loc->path);
        if (ret == -1) {
                op_errno = ENOMEM;
                gf_log (this->name, GF_LOG_WARNING,
                        "cannot remove parent entry from grand-parent's cache "
                        "for path (%s)", loc->path);
                goto out;
        }

        stub = fop_unlink_stub (frame, sp_unlink_helper, loc);
        if (stub == NULL) {
                op_errno = ENOMEM;
                goto out;
        }

        sp_process_inode_ctx (frame, this, loc, stub, &need_unwind,
                              &need_lookup, &can_wind, &op_errno);

out:
        if (need_unwind) {
                SP_STACK_UNWIND (unlink, frame, -1, op_errno, NULL, NULL);
        } else if (need_lookup) {
                STACK_WIND (frame, sp_lookup_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->lookup, loc, NULL);
        } else if (can_wind) {
                STACK_WIND (frame, sp_unlink_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->unlink, loc);
        }

        return 0;
}


int
sp_rmdir_helper (call_frame_t *frame, xlator_t *this, loc_t *loc, int flags)
{
        uint64_t        value     = 0;
        sp_inode_ctx_t *inode_ctx = NULL;
        int32_t         ret       = 0, op_ret = -1, op_errno = EINVAL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, loc, unwind);

        ret = inode_ctx_get (loc->inode, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "stat-prefetch context not set in inode "
                        "(ino:%"PRId64" gfid:%s)", loc->inode->ino,
                        uuid_utoa (loc->inode->gfid));
                goto unwind;
        }

        inode_ctx = (sp_inode_ctx_t *)(long) value;
        GF_VALIDATE_OR_GOTO (this->name, inode_ctx, unwind);

        LOCK (&inode_ctx->lock);
        {
                op_ret = inode_ctx->op_ret;
                op_errno = inode_ctx->op_errno;
        }
        UNLOCK (&inode_ctx->lock);

        if (op_ret == -1) {
                gf_log (this->name, GF_LOG_WARNING, "lookup-behind has failed "
                        "for path (%s)(%s), unwinding rmdir call "
                        "waiting on it", loc->path, strerror (op_errno));
                goto unwind;
        }

        STACK_WIND (frame, sp_unlink_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->rmdir, loc, flags);

        return 0;

unwind:
        SP_STACK_UNWIND (rmdir, frame, -1, op_errno, NULL, NULL);
        return 0;
}


int
sp_rmdir (call_frame_t *frame, xlator_t *this, loc_t *loc, int flags)
{
        int32_t         ret          = -1, op_errno = EINVAL;
        call_stub_t    *stub         = NULL;
        char            can_wind     = 0, need_lookup = 0, need_unwind = 1;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, out);
        GF_VALIDATE_OR_GOTO (this->name, loc, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->name, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->path, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->inode, out);

        sp_remove_caches_from_all_fds_opened (this, loc->inode, NULL);

        ret = sp_cache_remove_parent_entry (frame, this, loc->inode->table,
                                            (char *)loc->path);
        if (ret == -1) {
                op_errno = ENOMEM;
                gf_log (this->name, GF_LOG_WARNING,
                        "cannot remove parent entry from grand-parent's cache "
                        "for path (%s)", loc->path);
                goto out;
        }

        stub = fop_rmdir_stub (frame, sp_rmdir_helper, loc, flags);
        if (stub == NULL) {
                op_errno = ENOMEM;
                goto out;
        }

        sp_process_inode_ctx (frame, this, loc, stub, &need_unwind,
                              &need_lookup, &can_wind, &op_errno);

out:
        if (need_unwind) {
                SP_STACK_UNWIND (rmdir, frame, -1, op_errno, NULL, NULL);
        } else if (need_lookup) {
                STACK_WIND (frame, sp_lookup_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->lookup, loc, NULL);
        } else if (can_wind) {
                STACK_WIND (frame, sp_unlink_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->rmdir, loc, flags);
        }

        return 0;
}


int32_t
sp_readv_cbk (call_frame_t *frame, void *cookie, xlator_t *this, int32_t op_ret,
              int32_t op_errno, struct iovec *vector, int32_t count,
              struct iatt *stbuf, struct iobref *iobref)
{
        GF_ASSERT (frame);
        SP_STACK_UNWIND (readv, frame, op_ret, op_errno, vector, count, stbuf,
                         iobref);
        return 0;
}


int32_t
sp_readv (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
          off_t offset)
{
        sp_fd_ctx_t *fd_ctx = NULL;
        uint64_t     value  = 0;
        int32_t      ret    = 0, op_errno = EINVAL;
        inode_t     *parent = NULL;
        char        *name   = NULL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, fd, unwind);

        ret = fd_ctx_get (fd, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING, "stat-prefetch context not "
                        "set in fd (%p) opened on inode (ino:%"PRId64", "
                        "gfid:%s", fd, fd->inode->ino,
                        uuid_utoa (fd->inode->gfid));
                goto unwind;
        }

        fd_ctx = (void *)(long)value;
        name   = fd_ctx->name;
        parent = fd_ctx->parent_inode;

        sp_remove_caches_from_all_fds_opened (this, parent, (char *)name);

        STACK_WIND (frame, sp_readv_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->readv, fd, size, offset);
        return 0;

unwind:
        SP_STACK_UNWIND (readv, frame, -1, op_errno, NULL, -1, NULL, NULL);
        return 0;
}


int32_t
sp_writev (call_frame_t *frame, xlator_t *this, fd_t *fd, struct iovec *vector,
           int32_t count, off_t off, struct iobref *iobref)
{
        sp_fd_ctx_t *fd_ctx = NULL;
        uint64_t     value  = 0;
        int32_t      ret    = 0, op_errno = EINVAL;
        inode_t     *parent = NULL;
        char        *name   = NULL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, fd, unwind);

        ret = fd_ctx_get (fd, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING, "stat-prefetch context not "
                        "set in fd (%p) opened on inode (ino:%"PRId64", "
                        "gfid:%s", fd, fd->inode->ino,
                        uuid_utoa (fd->inode->gfid));
                goto unwind;
        }

        fd_ctx = (void *)(long)value;
        name   = fd_ctx->name;
        parent = fd_ctx->parent_inode;

        sp_remove_caches_from_all_fds_opened (this, parent, (char *)name);

        STACK_WIND (frame, sp_unlink_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->writev, fd, vector, count, off,
                    iobref);
        return 0;

unwind:
        SP_STACK_UNWIND (writev, frame, -1, op_errno, NULL, NULL);
        return 0;
}


int32_t
sp_fsync (call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t flags)
{
        sp_fd_ctx_t *fd_ctx = NULL;
        uint64_t     value  = 0;
        int32_t      ret    = 0, op_errno = EINVAL;
        inode_t     *parent = NULL;
        char        *name   = NULL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, fd, unwind);

        ret = fd_ctx_get (fd, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING, "stat-prefetch context not "
                        "set in fd (%p) opened on inode (ino:%"PRId64", "
                        "gfid:%s", fd, fd->inode->ino,
                        uuid_utoa (fd->inode->gfid));
                goto unwind;
        }

        fd_ctx = (void *)(long)value;
        name   = fd_ctx->name;
        parent = fd_ctx->parent_inode;

        sp_remove_caches_from_all_fds_opened (this, parent, (char *)name);

        STACK_WIND (frame, sp_unlink_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->fsync, fd, flags);
        return 0;

unwind:
        SP_STACK_UNWIND (fsync, frame, -1, op_errno, NULL, NULL);
        return 0;
}


int32_t
sp_rename_helper (call_frame_t *frame, xlator_t *this, loc_t *oldloc,
                  loc_t *newloc)
{
        uint64_t        value               = 0;
        char            need_unwind         = 0;
        char            can_wind            = 0;
        int32_t         ret                 = 0, op_errno = EINVAL;
        int32_t         old_op_ret          = -1, old_op_errno = -1;
        int32_t         new_op_ret          = -1, new_op_errno = -1;
        char            old_inode_looked_up = 0, new_inode_looked_up = 0;
        sp_inode_ctx_t *old_inode_ctx       = NULL, *new_inode_ctx = NULL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this ? frame->this->name : "stat-prefetch",
                             this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, oldloc, unwind);
        GF_VALIDATE_OR_GOTO (this->name, newloc, unwind);

        ret = inode_ctx_get (oldloc->inode, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "stat-prefetch context not set in inode "
                        "(ino:%"PRId64" gfid:%s)", oldloc->inode->ino,
                        uuid_utoa (oldloc->inode->gfid));
                goto unwind;
        }

        old_inode_ctx = (sp_inode_ctx_t *)(long) value;
        GF_VALIDATE_OR_GOTO (this->name, old_inode_ctx, unwind);

        LOCK (&old_inode_ctx->lock);
        {
                old_inode_looked_up = old_inode_ctx->looked_up;
                old_op_ret = old_inode_ctx->op_ret;
                old_op_errno = old_inode_ctx->op_errno;
                need_unwind = old_inode_ctx->need_unwind;
        }
        UNLOCK (&old_inode_ctx->lock);

        if (need_unwind) {
                /* there was an error while queuing up lookup stub for newloc */
                gf_log (this->name, GF_LOG_WARNING,
                        "could not queue lookup stub for path (%s)",
                        newloc->path);
                goto unwind;
        }

        if (newloc->inode != NULL) {
                ret = inode_ctx_get (newloc->inode, this, &value);
                if (ret == 0) {
                        new_inode_ctx = (sp_inode_ctx_t *)(long)value;
                        if (new_inode_ctx != NULL) {
                                LOCK (&new_inode_ctx->lock);
                                {
                                        new_inode_looked_up
                                                = new_inode_ctx->looked_up;
                                        new_op_ret = new_inode_ctx->op_ret;
                                        new_op_errno = new_inode_ctx->op_errno;
                                }
                                UNLOCK (&new_inode_ctx->lock);
                        }
                }
        }

        if (new_inode_ctx == NULL) {
                if (old_op_ret == -1) {
                        gf_log (this->name, GF_LOG_WARNING,
                                "lookup-behind has failed "
                                "for path (%s)(%s), unwinding rename call "
                                "waiting on it", oldloc->path,
                                strerror (old_op_errno));

                        op_errno = old_op_errno;
                        goto unwind;
                } else {
                        can_wind = 1;
                }
        } else {
                if (new_inode_looked_up && old_inode_looked_up) {
                        if ((old_op_ret == -1)
                            || ((new_op_ret == -1)
                                && (new_op_errno != ENOENT))) {
                                if (old_op_ret == -1) {
                                        gf_log (this->name, GF_LOG_WARNING,
                                                "lookup-behind has failed "
                                                "for path (%s)(%s), unwinding "
                                                "rename call waiting on it",
                                                oldloc->path,
                                                strerror (old_op_errno));
                                        op_errno = old_op_errno;
                                } else {
                                        gf_log (this->name, GF_LOG_WARNING,
                                                "lookup-behind has failed "
                                                "for path (%s)(%s), unwinding "
                                                "rename call waiting on it",
                                                newloc->path,
                                                strerror (new_op_errno));
                                        op_errno = new_op_errno;
                                }

                                goto unwind;
                        } else {
                                can_wind = 1;
                        }
                }
        }

        if (can_wind) {
                STACK_WIND (frame, sp_rename_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->rename, oldloc, newloc);
        }

        return 0;

unwind:
        SP_STACK_UNWIND (rename, frame, -1, op_errno, NULL, NULL, NULL, NULL,
                         NULL);
        return 0;
}


int32_t
sp_rename (call_frame_t *frame, xlator_t *this, loc_t *oldloc,loc_t *newloc)
{
        char            need_unwind           = 1;
        uint64_t        value                 = 0;
        call_stub_t    *stub                  = NULL;
        sp_inode_ctx_t *inode_ctx             = NULL;
        int32_t         ret                   = -1, op_errno = EINVAL;
        char            old_inode_can_wind    = 0, new_inode_can_wind = 0;
        char            old_inode_need_lookup = 0, new_inode_need_lookup = 0;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (frame->this ? frame->this->name : "stat-prefetch",
                             this, out);
        GF_VALIDATE_OR_GOTO (this->name, oldloc, out);
        GF_VALIDATE_OR_GOTO (this->name, oldloc->path, out);
        GF_VALIDATE_OR_GOTO (this->name, oldloc->name, out);
        GF_VALIDATE_OR_GOTO (this->name, oldloc->inode, out);

        GF_VALIDATE_OR_GOTO (this->name, newloc, out);
        GF_VALIDATE_OR_GOTO (this->name, newloc->path, out);

        sp_remove_caches_from_all_fds_opened (this, oldloc->parent,
                                              (char *)oldloc->name);

        sp_remove_caches_from_all_fds_opened (this, newloc->parent,
                                              (char *)newloc->name);

        ret = sp_cache_remove_parent_entry (frame, this, oldloc->parent->table,
                                            (char *)oldloc->path);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "cannot remove parent entry from grand-parent's cache "
                        "for path (%s)", oldloc->path);
                goto out;
        }

        ret = sp_cache_remove_parent_entry (frame, this, newloc->parent->table,
                                            (char *)newloc->path);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "cannot remove parent entry from grand-parent's cache "
                        "for path (%s)", newloc->path);
                goto out;
        }

        if (IA_ISDIR (oldloc->inode->ia_type)) {
                sp_remove_caches_from_all_fds_opened (this, oldloc->inode,
                                                      NULL);
        }

        stub = fop_rename_stub (frame, sp_rename_helper, oldloc, newloc);
        if (stub == NULL) {
                op_errno = ENOMEM;
                goto out;
        }

        ret = sp_process_inode_ctx (frame, this, oldloc, stub, &need_unwind,
                                    &old_inode_need_lookup, &old_inode_can_wind,
                                    &op_errno);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING, "processing stat-prefetch "
                        "context in inode (ino:%"PRId64", gfid:%s) (path:%s) "
                        "failed (%s)", oldloc->inode->ino,
                        uuid_utoa (oldloc->inode->gfid), oldloc->path,
                        strerror (op_errno));
                goto out;
        }

        if (newloc->inode != NULL) {
                stub = fop_rename_stub (frame, sp_rename_helper, oldloc,
                                        newloc);
                if (stub == NULL) {
                        op_errno = ENOMEM;
                        goto out;
                }

                ret = sp_process_inode_ctx (frame, this, newloc, stub,
                                            &need_unwind,
                                            &new_inode_need_lookup,
                                            &new_inode_can_wind, &op_errno);
                if (ret == -1) {
                        ret = inode_ctx_get (oldloc->inode, this, &value);

                        inode_ctx = (sp_inode_ctx_t *)(long)value;
                        if (inode_ctx == NULL) {
                                gf_log (this->name, GF_LOG_WARNING,
                                        "stat-prefetch context not set in inode"
                                        " (ino:%"PRId64", gfid:%s) (path:%s)",
                                        oldloc->inode->ino,
                                        uuid_utoa (oldloc->inode->gfid),
                                        oldloc->path);
                                goto out;
                        }

                        LOCK (&inode_ctx->lock);
                        {
                                if (!inode_ctx->looked_up) {
                                        /* unwind in sp_rename_helper */
                                        need_unwind = 0;
                                        inode_ctx->need_unwind = 1;
                                }
                        }
                        UNLOCK (&inode_ctx->lock);
                }

        } else {
                new_inode_can_wind = 1;
        }

out:
        if (need_unwind) {
                SP_STACK_UNWIND (rename, frame, -1, op_errno, NULL, NULL, NULL,
                                 NULL, NULL);
        } else if (old_inode_need_lookup || new_inode_need_lookup) {
                if (old_inode_need_lookup) {
                        STACK_WIND (frame, sp_lookup_cbk, FIRST_CHILD(this),
                                    FIRST_CHILD(this)->fops->lookup, oldloc,
                                    NULL);
                }

                if (new_inode_need_lookup) {
                        STACK_WIND (frame, sp_lookup_cbk, FIRST_CHILD(this),
                                    FIRST_CHILD(this)->fops->lookup, newloc,
                                    NULL);
                }
        } else if (old_inode_can_wind && new_inode_can_wind) {
                STACK_WIND (frame, sp_rename_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->rename, oldloc, newloc);
        }

        return 0;
}


int32_t
sp_setxattr_helper (call_frame_t *frame, xlator_t *this, loc_t *loc,
                    dict_t *dict, int32_t flags)
{
        uint64_t        value     = 0;
        sp_inode_ctx_t *inode_ctx = NULL;
        int32_t         ret       = 0, op_ret = -1, op_errno = EINVAL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, loc, unwind);

        ret = inode_ctx_get (loc->inode, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "stat-prefetch context not set in inode "
                        "(ino:%"PRId64" gfid:%s)", loc->inode->ino,
                        uuid_utoa (loc->inode->gfid));
                goto unwind;
        }

        inode_ctx = (sp_inode_ctx_t *)(long) value;
        GF_VALIDATE_OR_GOTO (this->name, inode_ctx, unwind);

        LOCK (&inode_ctx->lock);
        {
                op_ret = inode_ctx->op_ret;
                op_errno = inode_ctx->op_errno;
        }
        UNLOCK (&inode_ctx->lock);

        if (op_ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "lookup-behind has failed for path (%s)(%s), "
                        "unwinding setxattr call waiting on it", loc->path,
                        strerror (op_errno));
                goto unwind;
        }

        STACK_WIND (frame, sp_err_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->setxattr, loc, dict,
                    flags);

        return 0;

unwind:
        SP_STACK_UNWIND (setxattr, frame, -1, op_errno);
        return 0;
}


int32_t
sp_setxattr (call_frame_t *frame, xlator_t *this, loc_t *loc, dict_t *dict,
             int32_t flags)
{
        int32_t         op_errno     = EINVAL;
        call_stub_t    *stub         = NULL;
        char            can_wind     = 0, need_lookup = 0, need_unwind = 1;

        GF_VALIDATE_OR_GOTO (this->name, loc, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->name, out);

        sp_remove_caches_from_all_fds_opened (this, loc->parent,
                                              (char *)loc->name);

        stub = fop_setxattr_stub (frame, sp_setxattr_helper, loc, dict, flags);
        if (stub == NULL) {
                op_errno = ENOMEM;
                goto out;
        }

        sp_process_inode_ctx (frame, this, loc, stub, &need_unwind,
                              &need_lookup, &can_wind, &op_errno);

out:
        if (need_unwind) {
                SP_STACK_UNWIND (setxattr, frame, -1, op_errno);
        } else if (need_lookup) {
                STACK_WIND (frame, sp_lookup_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->lookup, loc, NULL);
        } else if (can_wind) {
                STACK_WIND (frame, sp_err_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->setxattr, loc, dict,
                            flags);
        }

        return 0;
}


int32_t
sp_removexattr_helper (call_frame_t *frame, xlator_t *this, loc_t *loc,
                       const char *name)
{
        uint64_t        value     = 0;
        sp_inode_ctx_t *inode_ctx = NULL;
        int32_t         ret       = 0, op_ret = -1, op_errno = EINVAL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, loc, unwind);

        ret = inode_ctx_get (loc->inode, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "stat-prefetch context not set in inode "
                        "(ino:%"PRId64" gfid:%s)", loc->inode->ino,
                        uuid_utoa (loc->inode->gfid));
                goto unwind;
        }

        inode_ctx = (sp_inode_ctx_t *)(long) value;
        GF_VALIDATE_OR_GOTO (this->name, inode_ctx, unwind);

        LOCK (&inode_ctx->lock);
        {
                op_ret = inode_ctx->op_ret;
                op_errno = inode_ctx->op_errno;
        }
        UNLOCK (&inode_ctx->lock);

        if (op_ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "lookup-behind has failed for path (%s)(%s), "
                        "unwinding setxattr call waiting on it", loc->path,
                        strerror (op_errno));
                goto unwind;
        }

        STACK_WIND (frame, sp_err_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->removexattr, loc, name);

        return 0;

unwind:
        SP_STACK_UNWIND (removexattr, frame, -1, op_errno);
        return 0;
}


int32_t
sp_removexattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
                const char *name)
{
        int32_t         op_errno     = EINVAL;
        call_stub_t    *stub         = NULL;
        char            can_wind     = 0, need_lookup = 0, need_unwind = 1;

        GF_VALIDATE_OR_GOTO (this->name, loc, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->name, out);

        sp_remove_caches_from_all_fds_opened (this, loc->parent,
                                              (char *)loc->name);

        stub = fop_removexattr_stub (frame, sp_removexattr_helper, loc, name);
        if (stub == NULL) {
                op_errno = ENOMEM;
                goto out;
        }

        sp_process_inode_ctx (frame, this, loc, stub, &need_unwind,
                              &need_lookup, &can_wind, &op_errno);

out:
        if (need_unwind) {
                SP_STACK_UNWIND (removexattr, frame, -1, op_errno);
        } else if (need_lookup) {
                STACK_WIND (frame, sp_lookup_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->lookup, loc, NULL);
        } else if (can_wind) {
                STACK_WIND (frame, sp_err_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->removexattr, loc, name);
        }

        return 0;
}


int32_t
sp_getxattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 int32_t op_ret, int32_t op_errno, dict_t *dict)
{
        GF_ASSERT (frame);
        SP_STACK_UNWIND (getxattr, frame, op_ret, op_errno, dict);
        return 0;
}


int32_t
sp_getxattr_helper (call_frame_t *frame, xlator_t *this, loc_t *loc,
                    const char *name)
{
        uint64_t        value     = 0;
        sp_inode_ctx_t *inode_ctx = NULL;
        int32_t         ret       = 0, op_ret = -1, op_errno = EINVAL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, loc, unwind);

        ret = inode_ctx_get (loc->inode, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "stat-prefetch context not set in inode "
                        "(ino:%"PRId64" gfid:%s)", loc->inode->ino,
                        uuid_utoa (loc->inode->gfid));
                goto unwind;
        }

        inode_ctx = (sp_inode_ctx_t *)(long) value;
        GF_VALIDATE_OR_GOTO (this->name, inode_ctx, unwind);

        LOCK (&inode_ctx->lock);
        {
                op_ret = inode_ctx->op_ret;
                op_errno = inode_ctx->op_errno;
        }
        UNLOCK (&inode_ctx->lock);

        if (op_ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "lookup-behind has failed for path (%s)(%s), "
                        "unwinding getxattr call waiting on it", loc->path,
                        strerror (op_errno));
                goto unwind;
        }

        STACK_WIND (frame, sp_getxattr_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->getxattr, loc, name);

        return 0;

unwind:
        SP_STACK_UNWIND (getxattr, frame, -1, op_errno, NULL);
        return 0;
}


int32_t
sp_getxattr (call_frame_t *frame, xlator_t *this, loc_t *loc, const char *name)
{
        int32_t         op_errno     = EINVAL;
        call_stub_t    *stub         = NULL;
        char            can_wind     = 0, need_lookup = 0, need_unwind = 1;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO (this->name, loc, out);
        GF_VALIDATE_OR_GOTO (this->name, loc->inode, out);

        stub = fop_getxattr_stub (frame, sp_getxattr_helper, loc, name);
        if (stub == NULL) {
                op_errno = ENOMEM;
                goto out;
        }

        sp_process_inode_ctx (frame, this, loc, stub, &need_unwind,
                              &need_lookup, &can_wind, &op_errno);

out:
        if (need_unwind) {
                SP_STACK_UNWIND (getxattr, frame, -1, op_errno, NULL);
        } else if (need_lookup) {
                STACK_WIND (frame, sp_lookup_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->lookup, loc, NULL);
        } else if (can_wind) {
                STACK_WIND (frame, sp_getxattr_cbk, FIRST_CHILD(this),
                            FIRST_CHILD(this)->fops->getxattr, loc, name);
        }

        return 0;
}


int32_t
sp_xattrop_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                int32_t op_ret, int32_t op_errno, dict_t *dict)
{
        GF_ASSERT (frame);
        SP_STACK_UNWIND (xattrop, frame, op_ret, op_errno, dict);
        return 0;
}


int32_t
sp_xattrop_helper (call_frame_t *frame, xlator_t *this, loc_t *loc,
                   gf_xattrop_flags_t flags, dict_t *dict)
{
        uint64_t        value     = 0;
        sp_inode_ctx_t *inode_ctx = NULL;
        int32_t         ret       = 0, op_ret = -1, op_errno = EINVAL;

        GF_ASSERT (frame);
        GF_VALIDATE_OR_GOTO ("stat-prefetch", this, unwind);
        GF_VALIDATE_OR_GOTO (this->name, loc, unwind);

        ret = inode_ctx_get (loc->inode, this, &value);
        if (ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "stat-prefetch context not set in inode "
                        "(ino:%"PRId64" gfid:%s)", loc->inode->ino,
                        uuid_utoa (loc->inode->gfid));
                goto unwind;
        }

        inode_ctx = (sp_inode_ctx_t *)(long) value;
        GF_VALIDATE_OR_GOTO (this->name, inode_ctx, unwind);

        LOCK (&inode_ctx->lock);
        {
                op_ret = inode_ctx->op_ret;
                op_errno = inode_ctx->op_errno;
        }
        UNLOCK (&inode_ctx->lock);

        if (op_ret == -1) {
                gf_log (this->name, GF_LOG_WARNING,
                        "lookup-behind has failed for path (%s)(%s), "
                        "unwinding xattrop call waiting on it", loc->path,
                        strerror (op_errno));
                goto unwind;
        }

        STACK_WIND (frame, sp_xattrop_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->xattrop, loc, flags, dict);

        return 0;