<commit_before>/*
  Copyright (c) 2006, 2007, 2008 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 <unistd.h>
#include <fcntl.h>
#include <limits.h>
#include <pthread.h>

#ifndef _CONFIG_H
#define _CONFIG_H
#include "config.h"
#endif

#include "glusterfs.h"
#include "compat.h"
#include "xlator.h"
#include "inode.h"
#include "logging.h"
#include "common-utils.h"

#include "locks.h"
#include "common.h"
#include "statedump.h"

#ifndef LLONG_MAX
#define LLONG_MAX LONG_LONG_MAX /* compat with old gcc */
#endif /* LLONG_MAX */

/* Forward declarations */


void do_blocked_rw (pl_inode_t *);
static int __rw_allowable (pl_inode_t *, posix_lock_t *, glusterfs_fop_t);

struct _truncate_ops {
        loc_t  loc;
        fd_t  *fd;
        off_t  offset;
        enum {TRUNCATE, FTRUNCATE} op;
};


int
pl_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)
{
        struct _truncate_ops *local = NULL;

        local = frame->local;

        if (local->op == TRUNCATE)
                loc_wipe (&local->loc);

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


static int
truncate_allowed (pl_inode_t *pl_inode,
                  void *transport, pid_t client_pid,
                  uint64_t owner, off_t offset)
{
        posix_lock_t *l = NULL;
        posix_lock_t  region = {.list = {0, }, };
        int           ret = 1;

        region.fl_start   = offset;
        region.fl_end     = LLONG_MAX;
        region.transport  = transport;
        region.client_pid = client_pid;
        region.owner      = owner;

        pthread_mutex_lock (&pl_inode->mutex);
        {
                list_for_each_entry (l, &pl_inode->ext_list, list) {
                        if (!l->blocked
                            && locks_overlap (&region, l)
                            && !same_owner (&region, l)) {
                                ret = 0;
                                break;
                        }
                }
        }
        pthread_mutex_unlock (&pl_inode->mutex);

        return ret;
}


static int
truncate_stat_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                   int32_t op_ret, int32_t op_errno, struct iatt *buf)
{
        posix_locks_private_t *priv = NULL;
        struct _truncate_ops  *local = NULL;
        inode_t               *inode = NULL;
        pl_inode_t            *pl_inode = NULL;


        priv = this->private;
        local = frame->local;

        if (op_ret != 0) {
                gf_log (this->name, GF_LOG_ERROR,
                        "got error (errno=%d, stderror=%s) from child",
                        op_errno, strerror (op_errno));
                goto unwind;
        }

        if (local->op == TRUNCATE)
                inode = local->loc.inode;
        else
                inode = local->fd->inode;

        pl_inode = pl_inode_get (this, inode);
        if (!pl_inode) {
                gf_log (this->name, GF_LOG_ERROR,
                        "Out of memory.");
                op_ret   = -1;
                op_errno = ENOMEM;
                goto unwind;
        }

        if (priv->mandatory
            && pl_inode->mandatory
            && !truncate_allowed (pl_inode, frame->root->trans,
                                  frame->root->pid, frame->root->lk_owner,
                                  local->offset)) {
                op_ret   = -1;
                op_errno = EAGAIN;
                goto unwind;
        }

        switch (local->op) {
        case TRUNCATE:
                STACK_WIND (frame, pl_truncate_cbk, FIRST_CHILD (this),
                            FIRST_CHILD (this)->fops->truncate,
                            &local->loc, local->offset);
                break;
        case FTRUNCATE:
                STACK_WIND (frame, pl_truncate_cbk, FIRST_CHILD (this),
                            FIRST_CHILD (this)->fops->ftruncate,
                            local->fd, local->offset);
                break;
        }

        return 0;

unwind:
        if (local->op == TRUNCATE)
                loc_wipe (&local->loc);

        STACK_UNWIND_STRICT (truncate, frame, op_ret, op_errno, buf, NULL);
        return 0;
}


int
pl_truncate (call_frame_t *frame, xlator_t *this,
             loc_t *loc, off_t offset)
{
        struct _truncate_ops *local = NULL;

	local = GF_CALLOC (1, sizeof (struct _truncate_ops),
                           gf_locks_mt_truncate_ops);
        if (!local) {
                gf_log (this->name, GF_LOG_ERROR,
                        "Out of memory.");
                goto unwind;
        }

        local->op         = TRUNCATE;
        local->offset     = offset;
        loc_copy (&local->loc, loc);

        frame->local = local;

        STACK_WIND (frame, truncate_stat_cbk, FIRST_CHILD (this),
                    FIRST_CHILD (this)->fops->stat, loc);

        return 0;

unwind:
        STACK_UNWIND_STRICT (truncate, frame, -1, ENOMEM, NULL, NULL);

        return 0;
}


int
pl_ftruncate (call_frame_t *frame, xlator_t *this,
              fd_t *fd, off_t offset)
{
        struct _truncate_ops *local = NULL;

	local = GF_CALLOC (1, sizeof (struct _truncate_ops),
                           gf_locks_mt_truncate_ops);
        if (!local) {
                gf_log (this->name, GF_LOG_ERROR,
                        "Out of memory.");
                goto unwind;
        }

        local->op         = FTRUNCATE;
        local->offset     = offset;
        local->fd         = fd;

        frame->local = local;

        STACK_WIND (frame, truncate_stat_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->fstat, fd);
        return 0;

unwind:
        STACK_UNWIND_STRICT (ftruncate, frame, -1, ENOMEM, NULL, NULL);

        return 0;
}

static void
delete_locks_of_fd (xlator_t *this, pl_inode_t *pl_inode, fd_t *fd)
{
       posix_lock_t *tmp = NULL;
       posix_lock_t *l = NULL;

       struct list_head blocked_list;

       INIT_LIST_HEAD (&blocked_list);

       pthread_mutex_lock (&pl_inode->mutex);
       {

               list_for_each_entry_safe (l, tmp, &pl_inode->ext_list, list) {
                       if ((l->fd_num == fd_to_fdnum(fd))) {
                               if (l->blocked) {
                                       list_move_tail (&l->list, &blocked_list);
                                       continue;
                               }
                               __delete_lock (pl_inode, l);
                               __destroy_lock (l);
                       }
               }

       }
       pthread_mutex_unlock (&pl_inode->mutex);

       list_for_each_entry_safe (l, tmp, &blocked_list, list) {
               list_del_init(&l->list);
               STACK_UNWIND_STRICT (lk, l->frame, -1, EAGAIN, &l->user_flock);
               __destroy_lock (l);
       }

        grant_blocked_locks (this, pl_inode);

        do_blocked_rw (pl_inode);

}

static void
__delete_locks_of_owner (pl_inode_t *pl_inode,
                         void *transport, uint64_t owner)
{
        posix_lock_t *tmp = NULL;
        posix_lock_t *l = NULL;

        /* TODO: what if it is a blocked lock with pending l->frame */

        list_for_each_entry_safe (l, tmp, &pl_inode->ext_list, list) {
                if ((l->transport == transport)
                    && (l->owner == owner)) {
			gf_log ("posix-locks", GF_LOG_TRACE,
                                " Flushing lock"
				"%s (pid=%d) (lk-owner=%"PRIu64") %"PRId64" - %"PRId64" state: %s",
				l->fl_type == F_UNLCK ? "Unlock" : "Lock",
				l->client_pid,
                                l->owner,
				l->user_flock.l_start,
				l->user_flock.l_len,
                                l->blocked == 1 ? "Blocked" : "Active");

                        __delete_lock (pl_inode, l);
                        __destroy_lock (l);
                }
        }

        return;
}

int32_t
pl_opendir_cbk (call_frame_t *frame,
		     void *cookie,
		     xlator_t *this,
		     int32_t op_ret,
		     int32_t op_errno,
		     fd_t *fd)
{
        int dummy = 1;
        int ret = -1;

        if (op_ret < 0)
                goto unwind;

        ret = fd_ctx_set (fd, this, dummy);
        if (ret != 0)
                gf_log (this->name, GF_LOG_ERROR,
                        "setting context for fd=%p in locks failed.", fd);

unwind:
	STACK_UNWIND_STRICT (opendir,
                             frame,
                             op_ret,
                             op_errno,
                             fd);
	return 0;
}

int32_t 
pl_opendir (call_frame_t *frame, xlator_t *this,
	     loc_t *loc, fd_t *fd)
{
	STACK_WIND (frame,
		    pl_opendir_cbk,
		    FIRST_CHILD(this),
		    FIRST_CHILD(this)->fops->opendir,
		    loc, fd);
	return 0;

}

int
pl_flush_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
              int32_t op_ret, int32_t op_errno)
{
        STACK_UNWIND_STRICT (flush, frame, op_ret, op_errno);

        return 0;
}


int
pl_flush (call_frame_t *frame, xlator_t *this,
          fd_t *fd)
{
        posix_locks_private_t *priv     = NULL;
        pl_inode_t            *pl_inode = NULL;
        uint64_t              owner     = -1;

        priv = this->private;
        owner = frame->root->lk_owner;

        pl_inode = pl_inode_get (this, fd->inode);

        if (!pl_inode) {
                gf_log (this->name, GF_LOG_DEBUG, "Could not get inode.");
                STACK_UNWIND_STRICT (flush, frame, -1, EBADFD);
                return 0;
        }

        pl_trace_flush (this, frame, fd);

        if (owner == 0) {
                /* Handle special case when protocol/server sets lk-owner to zero.
                 * This usually happens due to a client disconnection. Hence, free
                 * all locks opened with this fd.
                 */
                gf_log (this->name, GF_LOG_TRACE,
			"Releasing all locks with fd %p", fd);
                delete_locks_of_fd (this, pl_inode, fd);
                goto wind;

        }
        pthread_mutex_lock (&pl_inode->mutex);
        {
                __delete_locks_of_owner (pl_inode, frame->root->trans,
                                         owner);
        }
        pthread_mutex_unlock (&pl_inode->mutex);

        grant_blocked_locks (this, pl_inode);

        do_blocked_rw (pl_inode);

wind:
        STACK_WIND (frame, pl_flush_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->flush, fd);
        return 0;
}


int
pl_open_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
             int32_t op_ret, int32_t op_errno, fd_t *fd)
{
        int dummy = 1;
        int ret = -1;

        if (op_ret < 0)
                goto unwind;

        ret = fd_ctx_set (fd, this, dummy);
        if (ret != 0)
                gf_log (this->name, GF_LOG_ERROR,
                        "setting context for fd=%p in locks failed.", fd);

unwind:
        STACK_UNWIND_STRICT (open, frame, op_ret, op_errno, fd);

        return 0;
}


int
pl_open (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
         fd_t *fd, int32_t wbflags)
{
        /* why isn't O_TRUNC being handled ? */
        STACK_WIND (frame, pl_open_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->open,
                    loc, flags & ~O_TRUNC, fd, wbflags);

        return 0;
}


int
pl_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)
{
        int dummy = 1;
        int ret = -1;

        if (op_ret < 0)
                goto unwind;

        ret = fd_ctx_set (fd, this, dummy);
        if (ret != 0)
                gf_log (this->name, GF_LOG_ERROR,
                        "setting context for fd=%p in locks failed.", fd);

unwind:
        STACK_UNWIND_STRICT (create, frame, op_ret, op_errno, fd, inode, buf,
                             preparent, postparent);

        return 0;
}


int
pl_create (call_frame_t *frame, xlator_t *this,
           loc_t *loc, int32_t flags, mode_t mode, fd_t *fd)
{
        STACK_WIND (frame, pl_create_cbk,
                    FIRST_CHILD (this), FIRST_CHILD (this)->fops->create,
                    loc, flags, mode, fd);
        return 0;
}


int
pl_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)
{
        STACK_UNWIND_STRICT (readv, frame, op_ret, op_errno,
                             vector, count, stbuf, iobref);

        return 0;
}

int
pl_writev_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
               int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
               struct iatt *postbuf)
{
        STACK_UNWIND_STRICT (writev, frame, op_ret, op_errno, prebuf, postbuf);

        return 0;
}


void
do_blocked_rw (pl_inode_t *pl_inode)
{
        struct list_head  wind_list;
        pl_rw_req_t      *rw = NULL;
        pl_rw_req_t      *tmp = NULL;

        INIT_LIST_HEAD (&wind_list);

        pthread_mutex_lock (&pl_inode->mutex);
        {
                list_for_each_entry_safe (rw, tmp, &pl_inode->rw_list, list) {
                        if (__rw_allowable (pl_inode, &rw->region,
                                            rw->stub->fop)) {
                                list_del_init (&rw->list);
                                list_add_tail (&rw->list, &wind_list);
                        }
                }
        }
        pthread_mutex_unlock (&pl_inode->mutex);

        list_for_each_entry_safe (rw, tmp, &wind_list, list) {
                list_del_init (&rw->list);
                call_resume (rw->stub);
                GF_FREE (rw);
        }

        return;
}


static int
__rw_allowable (pl_inode_t *pl_inode, posix_lock_t *region,
                glusterfs_fop_t op)
{
        posix_lock_t *l = NULL;
        int           ret = 1;

        list_for_each_entry (l, &pl_inode->ext_list, list) {
                if (locks_overlap (l, region) && !same_owner (l, region)) {
                        if ((op == GF_FOP_READ) && (l->fl_type != F_WRLCK))
                                continue;
                        ret = 0;
                        break;
                }
        }

        return ret;
}


int
pl_readv_cont (call_frame_t *frame, xlator_t *this,
               fd_t *fd, size_t size, off_t offset)
{
        STACK_WIND (frame, pl_readv_cbk,
                    FIRST_CHILD (this), FIRST_CHILD (this)->fops->readv,
                    fd, size, offset);

        return 0;
}


int
pl_readv (call_frame_t *frame, xlator_t *this,
          fd_t *fd, size_t size, off_t offset)
{
        posix_locks_private_t *priv = NULL;
        pl_inode_t            *pl_inode = NULL;
        pl_rw_req_t           *rw = NULL;
        posix_lock_t           region = {.list = {0, }, };
        int                    op_ret = 0;
        int                    op_errno = 0;
        char                   wind_needed = 1;


        priv = this->private;
        pl_inode = pl_inode_get (this, fd->inode);

        if (priv->mandatory && pl_inode->mandatory) {
                region.fl_start   = offset;
                region.fl_end     = offset + size - 1;
                region.transport  = frame->root->trans;
                region.fd_num     = fd_to_fdnum(fd);
                region.client_pid = frame->root->pid;
                region.owner      = frame->root->lk_owner;

                pthread_mutex_lock (&pl_inode->mutex);
                {
                        wind_needed = __rw_allowable (pl_inode, &region,
                                                      GF_FOP_READ);
                        if (wind_needed) {
                                goto unlock;
                        }

                        if (fd->flags & O_NONBLOCK) {
                                gf_log (this->name, GF_LOG_TRACE,
                                        "returning EAGAIN as fd is O_NONBLOCK");
                                op_errno = EAGAIN;
                                op_ret = -1;
                                goto unlock;
                        }

			rw = GF_CALLOC (1, sizeof (*rw),
                                        gf_locks_mt_pl_rw_req_t);
                        if (!rw) {
                                gf_log (this->name, GF_LOG_ERROR,
                                        "Out of memory.");
                                op_errno = ENOMEM;
                                op_ret = -1;
                                goto unlock;
                        }

                        rw->stub = fop_readv_stub (frame, pl_readv_cont,
                                                   fd, size, offset);
                        if (!rw->stub) {
                                gf_log (this->name, GF_LOG_ERROR,
                                        "Out of memory.");
                                op_errno = ENOMEM;
                                op_ret = -1;
                                GF_FREE (rw);
                                goto unlock;
                        }

                        rw->region = region;

                        list_add_tail (&rw->list, &pl_inode->rw_list);
                }
        unlock:
                pthread_mutex_unlock (&pl_inode->mutex);
        }


        if (wind_needed) {
                STACK_WIND (frame, pl_readv_cbk,
                            FIRST_CHILD (this), FIRST_CHILD (this)->fops->readv,
                            fd, size, offset);
        }

        if (op_ret == -1)
                STACK_UNWIND_STRICT (readv, frame, -1, op_errno,
                                     NULL, 0, NULL, NULL);

        return 0;
}


int
pl_writev_cont (call_frame_t *frame, xlator_t *this, fd_t *fd,
                struct iovec *vector, int count, off_t offset,
                struct iobref *iobref)
{
        STACK_WIND (frame, pl_writev_cbk,
                    FIRST_CHILD (this), FIRST_CHILD (this)->fops->writev,
                    fd, vector, count, offset, iobref);

        return 0;
}


int
pl_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
           struct iovec *vector, int32_t count, off_t offset,
           struct iobref *iobref)
{
        posix_locks_private_t *priv = NULL;
        pl_inode_t            *pl_inode = NULL;
        pl_rw_req_t           *rw = NULL;
        posix_lock_t           region = {.list = {0, }, };
        int                    op_ret = 0;
        int                    op_errno = 0;
        char                   wind_needed = 1;


        priv = this->private;
        pl_inode = pl_inode_get (this, fd->inode);

        if (priv->mandatory && pl_inode->mandatory) {
                region.fl_start   = offset;
                region.fl_end     = offset + iov_length (vector, count) - 1;
                region.transport  = frame->root->trans;
                region.fd_num     = fd_to_fdnum(fd);
                region.client_pid = frame->root->pid;
                region.owner      = frame->root->lk_owner;

                pthread_mutex_lock (&pl_inode->mutex);
                {
                        wind_needed = __rw_allowable (pl_inode, &region,
                                                      GF_FOP_WRITE);
                        if (wind_needed)
                                goto unlock;

                        if (fd->flags & O_NONBLOCK) {
                                gf_log (this->name, GF_LOG_TRACE,
                                        "returning EAGAIN because fd is "
                                        "O_NONBLOCK");
                                op_errno = EAGAIN;
                                op_ret = -1;
                                goto unlock;
                        }

			rw = GF_CALLOC (1, sizeof (*rw),
                                        gf_locks_mt_pl_rw_req_t);
                        if (!rw) {
                                gf_log (this->name, GF_LOG_ERROR,
                                        "Out of memory.");
                                op_errno = ENOMEM;
                                op_ret = -1;
                                goto unlock;
                        }

                        rw->stub = fop_writev_stub (frame, pl_writev_cont,
                                                    fd, vector, count, offset,
                                                    iobref);
                        if (!rw->stub) {
                                gf_log (this->name, GF_LOG_ERROR,
                                        "Out of memory.");
                                op_errno = ENOMEM;
                                op_ret = -1;
                                GF_FREE (rw);
                                goto unlock;
                        }

                        rw->region = region;

                        list_add_tail (&rw->list, &pl_inode->rw_list);
                }
        unlock:
                pthread_mutex_unlock (&pl_inode->mutex);
        }


        if (wind_needed)
                STACK_WIND (frame, pl_writev_cbk,
                            FIRST_CHILD (this), FIRST_CHILD (this)->fops->writev,
                            fd, vector, count, offset, iobref);

        if (op_ret == -1)
                STACK_UNWIND_STRICT (writev, frame, -1, op_errno, NULL, NULL);

        return 0;
}


int
pl_lk (call_frame_t *frame, xlator_t *this,
       fd_t *fd, int32_t cmd, struct flock *flock)
{
        void                  *transport = NULL;
        pid_t                  client_pid = 0;
        uint64_t               owner      = 0;
        posix_locks_private_t *priv = NULL;
        pl_inode_t            *pl_inode = NULL;
        int                    op_ret = 0;
        int                    op_errno = 0;
        int                    can_block = 0;
        posix_lock_t          *reqlock = NULL;
        posix_lock_t          *conf = NULL;
        int                    ret = 0;

        transport  = frame->root->trans;
        client_pid = frame->root->pid;
        owner      = frame->root->lk_owner;
        priv       = this->private;

        pl_inode = pl_inode_get (this, fd->inode);
        if (!pl_inode) {
                gf_log (this->name, GF_LOG_ERROR,
                        "Out of memory.");
                op_ret = -1;
                op_errno = ENOMEM;
                goto unwind;
        }

        reqlock = new_posix_lock (flock, transport, client_pid,
                                  owner, fd);

        if (!reqlock) {
                gf_log (this->name, GF_LOG_ERROR,
                        "Out of memory.");
                op_ret = -1;
                op_errno = ENOMEM;
                goto unwind;
        }

        pl_trace_in (this, frame, fd, NULL, cmd, flock, NULL);

        switch (cmd) {

#if F_GETLK != F_GETLK64
        case F_GETLK64:
#endif
        case F_GETLK:
                conf = pl_getlk (pl_inode, reqlock);
                posix_lock_to_flock (conf, flock);
                __destroy_lock (reqlock);

                break;

#if F_SETLKW != F_SETLKW64
        case F_SETLKW64:
#endif
        case F_SETLKW:
                can_block = 1;
                reqlock->frame  = frame;
                reqlock->this   = this;

                /* fall through */

#if F_SETLK != F_SETLK64
        case F_SETLK64:
#endif
        case F_SETLK:
                memcpy (&reqlock->user_flock, flock, sizeof (struct flock));
                ret = pl_setlk (this, pl_inode, reqlock,
                                can_block);

                if (ret == -1) {
                        if (can_block) {
                                pl_trace_block (this, frame, fd, NULL, cmd, flock, NULL);
                                goto out;
                        }
                        gf_log (this->name, GF_LOG_DEBUG, "returning EAGAIN");
                        op_ret = -1;
                        op_errno = EAGAIN;
                        __destroy_lock (reqlock);
                }
        }

unwind:
        pl_trace_out (this, frame, fd, NULL, cmd, flock, op_ret, op_errno, NULL);
        pl_update_refkeeper (this, fd->inode);
        STACK_UNWIND_STRICT (lk, frame, op_ret, op_errno, flock);
out:
        return 0;
}


/* TODO: this function just logs, no action required?? */
int
pl_forget (xlator_t *this,
           inode_t *inode)
{
        pl_inode_t   *pl_inode = NULL;

        posix_lock_t *ext_tmp = NULL;
        posix_lock_t *ext_l   = NULL;
	struct list_head posixlks_released;

        pl_inode_lock_t *ino_tmp = NULL;
        pl_inode_lock_t *ino_l   = NULL;
	struct list_head inodelks_released;

        pl_rw_req_t *rw_tmp = NULL;
        pl_rw_req_t *rw_req = NULL;

        pl_entry_lock_t *entry_tmp = NULL;
        pl_entry_lock_t *entry_l   = NULL;
	struct list_head entrylks_released;

        pl_dom_list_t *dom = NULL;
        pl_dom_list_t *dom_tmp = NULL;

	INIT_LIST_HEAD (&posixlks_released);
	INIT_LIST_HEAD (&inodelks_released);
	INIT_LIST_HEAD (&entrylks_released);

        pl_inode = pl_inode_get (this, inode);

	pthread_mutex_lock (&pl_inode->mutex);
	{

		if (!list_empty (&pl_inode->rw_list)) {
			gf_log (this->name, GF_LOG_WARNING,
				"Pending R/W requests found, releasing.");

			list_for_each_entry_safe (rw_req, rw_tmp, &pl_inode->rw_list,
						  list) {

				list_del (&rw_req->list);
				GF_FREE (rw_req);
			}
		}

		if (!list_empty (&pl_inode->ext_list)) {
			gf_log (this->name, GF_LOG_WARNING,
				"Pending fcntl locks found, releasing.");

			list_for_each_entry_safe (ext_l, ext_tmp, &pl_inode->ext_list,
						  list) {

				__delete_lock (pl_inode, ext_l);
				if (ext_l->blocked) {
					list_add_tail (&ext_l->list, &posixlks_released);
					continue;
				}
				__destroy_lock (ext_l);
			}
		}


		list_for_each_entry_safe (dom, dom_tmp, &pl_inode->dom_list, inode_list) {

			if (!list_empty (&dom->inodelk_list)) {
				gf_log (this->name, GF_LOG_WARNING,
					"Pending inode locks found, releasing.");

				list_for_each_entry_safe (ino_l, ino_tmp, &dom->inodelk_list, list) {
					__delete_inode_lock (ino_l);
					__destroy_inode_lock (ino_l);
				}

				list_splice_init (&dom->blocked_inodelks, &inodelks_released);
				

			}
			if (!list_empty (&dom->entrylk_list)) {
				gf_log (this->name, GF_LOG_WARNING,
					"Pending entry locks found, releasing.");

				list_for_each_entry_safe (entry_l, entry_tmp, &dom->entrylk_list, domain_list) {
					list_del_init (&entry_l->domain_list);

					if (entry_l->basename)
						GF_FREE ((char *)entry_l->basename);
					GF_FREE (entry_l);
				}

				list_splice_init (&dom->blocked_entrylks, &entrylks_released);
			}

			list_del (&dom->inode_list);
			gf_log ("posix-locks", GF_LOG_TRACE,
				" Cleaning up domain: %s", dom->domain);
			GF_FREE ((char *)(dom->domain));
			GF_FREE (dom);
		}

	}
	pthread_mutex_unlock (&pl_inode->mutex);

	list_for_each_entry_safe (ext_l, ext_tmp, &posixlks_released, list) {

		STACK_UNWIND_STRICT (lk, ext_l->frame, -1, 0, &ext_l->user_flock);
		__destroy_lock (ext_l);
	}

	list_for_each_entry_safe (ino_l, ino_tmp, &inodelks_released, blocked_locks) {

		STACK_UNWIND_STRICT (inodelk, ino_l->frame, -1, 0);
		__destroy_inode_lock (ino_l);
	}

	list_for_each_entry_safe (entry_l, entry_tmp, &entrylks_released, blocked_locks) {

		STACK_UNWIND_STRICT (entrylk, entry_l->frame, -1, 0);
		if (entry_l->basename)
			GF_FREE ((char *)entry_l->basename);
		GF_FREE (entry_l);

	}

        GF_FREE (pl_inode);

        return 0;
}

int
pl_release (xlator_t *this, fd_t *fd)
{
        pl_inode_t *pl_inode     = NULL;
        uint64_t    tmp_pl_inode = 0;
        int         ret          = -1;

	ret = inode_ctx_get (fd->inode, this, &tmp_pl_inode);
        if (ret != 0)
                goto out;

        pl_inode = (pl_inode_t *)(long)tmp_pl_inode;

        pl_trace_release (this, fd);

        gf_log (this->name, GF_LOG_TRACE,
                "Releasing all locks with fd %p", fd);

        delete_locks_of_fd (this, pl_inode, fd);

out:
        return ret;
}
static int32_t
__get_posixlk_count (xlator_t *this, pl_inode_t *pl_inode)
{
        posix_lock_t *lock   = NULL;
        int32_t       count  = 0;

        list_for_each_entry (lock, &pl_inode->ext_list, list) {

			gf_log (this->name, GF_LOG_DEBUG,
                                " XATTR DEBUG"
				"%s (pid=%d) (lk-owner=%"PRIu64") %"PRId64" - %"PRId64" state: %s",
				lock->fl_type == F_UNLCK ? "Unlock" : "Lock",
				lock->client_pid,
                                lock->owner,
				lock->user_flock.l_start,
				lock->user_flock.l_len,
                                lock->blocked == 1 ? "Blocked" : "Active");

                count++;
        }

        return count;
}

int32_t
get_posixlk_count (xlator_t *this, inode_t *inode)
{
        pl_inode_t   *pl_inode = NULL;
        uint64_t      tmp_pl_inode = 0;
        int           ret      = 0;
        int32_t       count    = 0;

        ret = inode_ctx_get (inode, this, &tmp_pl_inode);
        if (ret != 0) {
                goto out;
        }

        pl_inode = (pl_inode_t *)(long) tmp_pl_inode;

        pthread_mutex_lock (&pl_inode->mutex);
        {
                count =__get_posixlk_count (this, pl_inode);
        }
        pthread_mutex_unlock (&pl_inode->mutex);

out:
        return count;
}

void
pl_entrylk_xattr_fill (xlator_t *this, inode_t *inode,
                       dict_t *dict)
{
        int32_t     count = 0;
        int         ret   = -1;

        count = get_entrylk_count (this, inode);
        ret = dict_set_int32 (dict, GLUSTERFS_ENTRYLK_COUNT, count);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_DEBUG,
                        " dict_set failed on key %s", GLUSTERFS_ENTRYLK_COUNT);
        }

}

void
pl_inodelk_xattr_fill (xlator_t *this, inode_t *inode,
                       dict_t *dict)
{
        int32_t     count = 0;
        int         ret   = -1;

        count = get_inodelk_count (this, inode);
        ret = dict_set_int32 (dict, GLUSTERFS_INODELK_COUNT, count);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_DEBUG,
                        " dict_set failed on key %s", GLUSTERFS_INODELK_COUNT);
        }

}

void
pl_posixlk_xattr_fill (xlator_t *this, inode_t *inode,
                       dict_t *dict)
{
        int32_t     count = 0;
        int         ret   = -1;

        count = get_posixlk_count (this, inode);
        ret = dict_set_int32 (dict, GLUSTERFS_POSIXLK_COUNT, count);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_DEBUG,
                        " dict_set failed on key %s", GLUSTERFS_POSIXLK_COUNT);
        }

}

int32_t
pl_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)
{
        pl_local_t *local = NULL;

        if (!frame->local) {
                goto out;
        }

        if (op_ret) {
                goto out;
        }

        local = frame->local;

        if (local->entrylk_count_req)
                pl_entrylk_xattr_fill (this, inode, dict);
        if (local->inodelk_count_req)
                pl_inodelk_xattr_fill (this, inode, dict);
        if (local->posixlk_count_req)
                pl_posixlk_xattr_fill (this, inode, dict);


        frame->local = NULL;

        if (local != NULL)
                GF_FREE (local);

out:
	STACK_UNWIND (frame,
		      op_ret,
		      op_errno,
		      inode,
		      buf,
                      dict,
                      postparent);<commit_msg>features/locks convert STACK_UNWIND to STACK_UNWIND_STRICT<commit_after>/*
  Copyright (c) 2006, 2007, 2008 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 <unistd.h>
#include <fcntl.h>
#include <limits.h>
#include <pthread.h>

#ifndef _CONFIG_H
#define _CONFIG_H
#include "config.h"
#endif

#include "glusterfs.h"
#include "compat.h"
#include "xlator.h"
#include "inode.h"
#include "logging.h"
#include "common-utils.h"

#include "locks.h"
#include "common.h"
#include "statedump.h"

#ifndef LLONG_MAX
#define LLONG_MAX LONG_LONG_MAX /* compat with old gcc */
#endif /* LLONG_MAX */

/* Forward declarations */


void do_blocked_rw (pl_inode_t *);
static int __rw_allowable (pl_inode_t *, posix_lock_t *, glusterfs_fop_t);

struct _truncate_ops {
        loc_t  loc;
        fd_t  *fd;
        off_t  offset;
        enum {TRUNCATE, FTRUNCATE} op;
};


int
pl_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)
{
        struct _truncate_ops *local = NULL;

        local = frame->local;

        if (local->op == TRUNCATE)
                loc_wipe (&local->loc);

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


static int
truncate_allowed (pl_inode_t *pl_inode,
                  void *transport, pid_t client_pid,
                  uint64_t owner, off_t offset)
{
        posix_lock_t *l = NULL;
        posix_lock_t  region = {.list = {0, }, };
        int           ret = 1;

        region.fl_start   = offset;
        region.fl_end     = LLONG_MAX;
        region.transport  = transport;
        region.client_pid = client_pid;
        region.owner      = owner;

        pthread_mutex_lock (&pl_inode->mutex);
        {
                list_for_each_entry (l, &pl_inode->ext_list, list) {
                        if (!l->blocked
                            && locks_overlap (&region, l)
                            && !same_owner (&region, l)) {
                                ret = 0;
                                break;
                        }
                }
        }
        pthread_mutex_unlock (&pl_inode->mutex);

        return ret;
}


static int
truncate_stat_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                   int32_t op_ret, int32_t op_errno, struct iatt *buf)
{
        posix_locks_private_t *priv = NULL;
        struct _truncate_ops  *local = NULL;
        inode_t               *inode = NULL;
        pl_inode_t            *pl_inode = NULL;


        priv = this->private;
        local = frame->local;

        if (op_ret != 0) {
                gf_log (this->name, GF_LOG_ERROR,
                        "got error (errno=%d, stderror=%s) from child",
                        op_errno, strerror (op_errno));
                goto unwind;
        }

        if (local->op == TRUNCATE)
                inode = local->loc.inode;
        else
                inode = local->fd->inode;

        pl_inode = pl_inode_get (this, inode);
        if (!pl_inode) {
                gf_log (this->name, GF_LOG_ERROR,
                        "Out of memory.");
                op_ret   = -1;
                op_errno = ENOMEM;
                goto unwind;
        }

        if (priv->mandatory
            && pl_inode->mandatory
            && !truncate_allowed (pl_inode, frame->root->trans,
                                  frame->root->pid, frame->root->lk_owner,
                                  local->offset)) {
                op_ret   = -1;
                op_errno = EAGAIN;
                goto unwind;
        }

        switch (local->op) {
        case TRUNCATE:
                STACK_WIND (frame, pl_truncate_cbk, FIRST_CHILD (this),
                            FIRST_CHILD (this)->fops->truncate,
                            &local->loc, local->offset);
                break;
        case FTRUNCATE:
                STACK_WIND (frame, pl_truncate_cbk, FIRST_CHILD (this),
                            FIRST_CHILD (this)->fops->ftruncate,
                            local->fd, local->offset);
                break;
        }

        return 0;

unwind:
        if (local->op == TRUNCATE)
                loc_wipe (&local->loc);

        STACK_UNWIND_STRICT (truncate, frame, op_ret, op_errno, buf, NULL);
        return 0;
}


int
pl_truncate (call_frame_t *frame, xlator_t *this,
             loc_t *loc, off_t offset)
{
        struct _truncate_ops *local = NULL;

	local = GF_CALLOC (1, sizeof (struct _truncate_ops),
                           gf_locks_mt_truncate_ops);
        if (!local) {
                gf_log (this->name, GF_LOG_ERROR,
                        "Out of memory.");
                goto unwind;
        }

        local->op         = TRUNCATE;
        local->offset     = offset;
        loc_copy (&local->loc, loc);

        frame->local = local;

        STACK_WIND (frame, truncate_stat_cbk, FIRST_CHILD (this),
                    FIRST_CHILD (this)->fops->stat, loc);

        return 0;

unwind:
        STACK_UNWIND_STRICT (truncate, frame, -1, ENOMEM, NULL, NULL);

        return 0;
}


int
pl_ftruncate (call_frame_t *frame, xlator_t *this,
              fd_t *fd, off_t offset)
{
        struct _truncate_ops *local = NULL;

	local = GF_CALLOC (1, sizeof (struct _truncate_ops),
                           gf_locks_mt_truncate_ops);
        if (!local) {
                gf_log (this->name, GF_LOG_ERROR,
                        "Out of memory.");
                goto unwind;
        }

        local->op         = FTRUNCATE;
        local->offset     = offset;
        local->fd         = fd;

        frame->local = local;

        STACK_WIND (frame, truncate_stat_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->fstat, fd);
        return 0;

unwind:
        STACK_UNWIND_STRICT (ftruncate, frame, -1, ENOMEM, NULL, NULL);

        return 0;
}

static void
delete_locks_of_fd (xlator_t *this, pl_inode_t *pl_inode, fd_t *fd)
{
       posix_lock_t *tmp = NULL;
       posix_lock_t *l = NULL;

       struct list_head blocked_list;

       INIT_LIST_HEAD (&blocked_list);

       pthread_mutex_lock (&pl_inode->mutex);
       {

               list_for_each_entry_safe (l, tmp, &pl_inode->ext_list, list) {
                       if ((l->fd_num == fd_to_fdnum(fd))) {
                               if (l->blocked) {
                                       list_move_tail (&l->list, &blocked_list);
                                       continue;
                               }
                               __delete_lock (pl_inode, l);
                               __destroy_lock (l);
                       }
               }

       }
       pthread_mutex_unlock (&pl_inode->mutex);

       list_for_each_entry_safe (l, tmp, &blocked_list, list) {
               list_del_init(&l->list);
               STACK_UNWIND_STRICT (lk, l->frame, -1, EAGAIN, &l->user_flock);
               __destroy_lock (l);
       }

        grant_blocked_locks (this, pl_inode);

        do_blocked_rw (pl_inode);

}

static void
__delete_locks_of_owner (pl_inode_t *pl_inode,
                         void *transport, uint64_t owner)
{
        posix_lock_t *tmp = NULL;
        posix_lock_t *l = NULL;

        /* TODO: what if it is a blocked lock with pending l->frame */

        list_for_each_entry_safe (l, tmp, &pl_inode->ext_list, list) {
                if ((l->transport == transport)
                    && (l->owner == owner)) {
			gf_log ("posix-locks", GF_LOG_TRACE,
                                " Flushing lock"
				"%s (pid=%d) (lk-owner=%"PRIu64") %"PRId64" - %"PRId64" state: %s",
				l->fl_type == F_UNLCK ? "Unlock" : "Lock",
				l->client_pid,
                                l->owner,
				l->user_flock.l_start,
				l->user_flock.l_len,
                                l->blocked == 1 ? "Blocked" : "Active");

                        __delete_lock (pl_inode, l);
                        __destroy_lock (l);
                }
        }

        return;
}

int32_t
pl_opendir_cbk (call_frame_t *frame,
		     void *cookie,
		     xlator_t *this,
		     int32_t op_ret,
		     int32_t op_errno,
		     fd_t *fd)
{
        int dummy = 1;
        int ret = -1;

        if (op_ret < 0)
                goto unwind;

        ret = fd_ctx_set (fd, this, dummy);
        if (ret != 0)
                gf_log (this->name, GF_LOG_ERROR,
                        "setting context for fd=%p in locks failed.", fd);

unwind:
	STACK_UNWIND_STRICT (opendir,
                             frame,
                             op_ret,
                             op_errno,
                             fd);
	return 0;
}

int32_t 
pl_opendir (call_frame_t *frame, xlator_t *this,
	     loc_t *loc, fd_t *fd)
{
	STACK_WIND (frame,
		    pl_opendir_cbk,
		    FIRST_CHILD(this),
		    FIRST_CHILD(this)->fops->opendir,
		    loc, fd);
	return 0;

}

int
pl_flush_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
              int32_t op_ret, int32_t op_errno)
{
        STACK_UNWIND_STRICT (flush, frame, op_ret, op_errno);

        return 0;
}


int
pl_flush (call_frame_t *frame, xlator_t *this,
          fd_t *fd)
{
        posix_locks_private_t *priv     = NULL;
        pl_inode_t            *pl_inode = NULL;
        uint64_t              owner     = -1;

        priv = this->private;
        owner = frame->root->lk_owner;

        pl_inode = pl_inode_get (this, fd->inode);

        if (!pl_inode) {
                gf_log (this->name, GF_LOG_DEBUG, "Could not get inode.");
                STACK_UNWIND_STRICT (flush, frame, -1, EBADFD);
                return 0;
        }

        pl_trace_flush (this, frame, fd);

        if (owner == 0) {
                /* Handle special case when protocol/server sets lk-owner to zero.
                 * This usually happens due to a client disconnection. Hence, free
                 * all locks opened with this fd.
                 */
                gf_log (this->name, GF_LOG_TRACE,
			"Releasing all locks with fd %p", fd);
                delete_locks_of_fd (this, pl_inode, fd);
                goto wind;

        }
        pthread_mutex_lock (&pl_inode->mutex);
        {
                __delete_locks_of_owner (pl_inode, frame->root->trans,
                                         owner);
        }
        pthread_mutex_unlock (&pl_inode->mutex);

        grant_blocked_locks (this, pl_inode);

        do_blocked_rw (pl_inode);

wind:
        STACK_WIND (frame, pl_flush_cbk, FIRST_CHILD(this),
                    FIRST_CHILD(this)->fops->flush, fd);
        return 0;
}


int
pl_open_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
             int32_t op_ret, int32_t op_errno, fd_t *fd)
{
        int dummy = 1;
        int ret = -1;

        if (op_ret < 0)
                goto unwind;

        ret = fd_ctx_set (fd, this, dummy);
        if (ret != 0)
                gf_log (this->name, GF_LOG_ERROR,
                        "setting context for fd=%p in locks failed.", fd);

unwind:
        STACK_UNWIND_STRICT (open, frame, op_ret, op_errno, fd);

        return 0;
}


int
pl_open (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
         fd_t *fd, int32_t wbflags)
{
        /* why isn't O_TRUNC being handled ? */
        STACK_WIND (frame, pl_open_cbk,
                    FIRST_CHILD(this), FIRST_CHILD(this)->fops->open,
                    loc, flags & ~O_TRUNC, fd, wbflags);

        return 0;
}


int
pl_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)
{
        int dummy = 1;
        int ret = -1;

        if (op_ret < 0)
                goto unwind;

        ret = fd_ctx_set (fd, this, dummy);
        if (ret != 0)
                gf_log (this->name, GF_LOG_ERROR,
                        "setting context for fd=%p in locks failed.", fd);

unwind:
        STACK_UNWIND_STRICT (create, frame, op_ret, op_errno, fd, inode, buf,
                             preparent, postparent);

        return 0;
}


int
pl_create (call_frame_t *frame, xlator_t *this,
           loc_t *loc, int32_t flags, mode_t mode, fd_t *fd)
{
        STACK_WIND (frame, pl_create_cbk,
                    FIRST_CHILD (this), FIRST_CHILD (this)->fops->create,
                    loc, flags, mode, fd);
        return 0;
}


int
pl_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)
{
        STACK_UNWIND_STRICT (readv, frame, op_ret, op_errno,
                             vector, count, stbuf, iobref);

        return 0;
}

int
pl_writev_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
               int32_t op_ret, int32_t op_errno, struct iatt *prebuf,
               struct iatt *postbuf)
{
        STACK_UNWIND_STRICT (writev, frame, op_ret, op_errno, prebuf, postbuf);

        return 0;
}


void
do_blocked_rw (pl_inode_t *pl_inode)
{
        struct list_head  wind_list;
        pl_rw_req_t      *rw = NULL;
        pl_rw_req_t      *tmp = NULL;

        INIT_LIST_HEAD (&wind_list);

        pthread_mutex_lock (&pl_inode->mutex);
        {
                list_for_each_entry_safe (rw, tmp, &pl_inode->rw_list, list) {
                        if (__rw_allowable (pl_inode, &rw->region,
                                            rw->stub->fop)) {
                                list_del_init (&rw->list);
                                list_add_tail (&rw->list, &wind_list);
                        }
                }
        }
        pthread_mutex_unlock (&pl_inode->mutex);

        list_for_each_entry_safe (rw, tmp, &wind_list, list) {
                list_del_init (&rw->list);
                call_resume (rw->stub);
                GF_FREE (rw);
        }

        return;
}


static int
__rw_allowable (pl_inode_t *pl_inode, posix_lock_t *region,
                glusterfs_fop_t op)
{
        posix_lock_t *l = NULL;
        int           ret = 1;

        list_for_each_entry (l, &pl_inode->ext_list, list) {
                if (locks_overlap (l, region) && !same_owner (l, region)) {
                        if ((op == GF_FOP_READ) && (l->fl_type != F_WRLCK))
                                continue;
                        ret = 0;
                        break;
                }
        }

        return ret;
}


int
pl_readv_cont (call_frame_t *frame, xlator_t *this,
               fd_t *fd, size_t size, off_t offset)
{
        STACK_WIND (frame, pl_readv_cbk,
                    FIRST_CHILD (this), FIRST_CHILD (this)->fops->readv,
                    fd, size, offset);

        return 0;
}


int
pl_readv (call_frame_t *frame, xlator_t *this,
          fd_t *fd, size_t size, off_t offset)
{
        posix_locks_private_t *priv = NULL;
        pl_inode_t            *pl_inode = NULL;
        pl_rw_req_t           *rw = NULL;
        posix_lock_t           region = {.list = {0, }, };
        int                    op_ret = 0;
        int                    op_errno = 0;
        char                   wind_needed = 1;


        priv = this->private;
        pl_inode = pl_inode_get (this, fd->inode);

        if (priv->mandatory && pl_inode->mandatory) {
                region.fl_start   = offset;
                region.fl_end     = offset + size - 1;
                region.transport  = frame->root->trans;
                region.fd_num     = fd_to_fdnum(fd);
                region.client_pid = frame->root->pid;
                region.owner      = frame->root->lk_owner;

                pthread_mutex_lock (&pl_inode->mutex);
                {
                        wind_needed = __rw_allowable (pl_inode, &region,
                                                      GF_FOP_READ);
                        if (wind_needed) {
                                goto unlock;
                        }

                        if (fd->flags & O_NONBLOCK) {
                                gf_log (this->name, GF_LOG_TRACE,
                                        "returning EAGAIN as fd is O_NONBLOCK");
                                op_errno = EAGAIN;
                                op_ret = -1;
                                goto unlock;
                        }

			rw = GF_CALLOC (1, sizeof (*rw),
                                        gf_locks_mt_pl_rw_req_t);
                        if (!rw) {
                                gf_log (this->name, GF_LOG_ERROR,
                                        "Out of memory.");
                                op_errno = ENOMEM;
                                op_ret = -1;
                                goto unlock;
                        }

                        rw->stub = fop_readv_stub (frame, pl_readv_cont,
                                                   fd, size, offset);
                        if (!rw->stub) {
                                gf_log (this->name, GF_LOG_ERROR,
                                        "Out of memory.");
                                op_errno = ENOMEM;
                                op_ret = -1;
                                GF_FREE (rw);
                                goto unlock;
                        }

                        rw->region = region;

                        list_add_tail (&rw->list, &pl_inode->rw_list);
                }
        unlock:
                pthread_mutex_unlock (&pl_inode->mutex);
        }


        if (wind_needed) {
                STACK_WIND (frame, pl_readv_cbk,
                            FIRST_CHILD (this), FIRST_CHILD (this)->fops->readv,
                            fd, size, offset);
        }

        if (op_ret == -1)
                STACK_UNWIND_STRICT (readv, frame, -1, op_errno,
                                     NULL, 0, NULL, NULL);

        return 0;
}


int
pl_writev_cont (call_frame_t *frame, xlator_t *this, fd_t *fd,
                struct iovec *vector, int count, off_t offset,
                struct iobref *iobref)
{
        STACK_WIND (frame, pl_writev_cbk,
                    FIRST_CHILD (this), FIRST_CHILD (this)->fops->writev,
                    fd, vector, count, offset, iobref);

        return 0;
}


int
pl_writev (call_frame_t *frame, xlator_t *this, fd_t *fd,
           struct iovec *vector, int32_t count, off_t offset,
           struct iobref *iobref)
{
        posix_locks_private_t *priv = NULL;
        pl_inode_t            *pl_inode = NULL;
        pl_rw_req_t           *rw = NULL;
        posix_lock_t           region = {.list = {0, }, };
        int                    op_ret = 0;
        int                    op_errno = 0;
        char                   wind_needed = 1;


        priv = this->private;
        pl_inode = pl_inode_get (this, fd->inode);

        if (priv->mandatory && pl_inode->mandatory) {
                region.fl_start   = offset;
                region.fl_end     = offset + iov_length (vector, count) - 1;
                region.transport  = frame->root->trans;
                region.fd_num     = fd_to_fdnum(fd);
                region.client_pid = frame->root->pid;
                region.owner      = frame->root->lk_owner;

                pthread_mutex_lock (&pl_inode->mutex);
                {
                        wind_needed = __rw_allowable (pl_inode, &region,
                                                      GF_FOP_WRITE);
                        if (wind_needed)
                                goto unlock;

                        if (fd->flags & O_NONBLOCK) {
                                gf_log (this->name, GF_LOG_TRACE,
                                        "returning EAGAIN because fd is "
                                        "O_NONBLOCK");
                                op_errno = EAGAIN;
                                op_ret = -1;
                                goto unlock;
                        }

			rw = GF_CALLOC (1, sizeof (*rw),
                                        gf_locks_mt_pl_rw_req_t);
                        if (!rw) {
                                gf_log (this->name, GF_LOG_ERROR,
                                        "Out of memory.");
                                op_errno = ENOMEM;
                                op_ret = -1;
                                goto unlock;
                        }

                        rw->stub = fop_writev_stub (frame, pl_writev_cont,
                                                    fd, vector, count, offset,
                                                    iobref);
                        if (!rw->stub) {
                                gf_log (this->name, GF_LOG_ERROR,
                                        "Out of memory.");
                                op_errno = ENOMEM;
                                op_ret = -1;
                                GF_FREE (rw);
                                goto unlock;
                        }

                        rw->region = region;

                        list_add_tail (&rw->list, &pl_inode->rw_list);
                }
        unlock:
                pthread_mutex_unlock (&pl_inode->mutex);
        }


        if (wind_needed)
                STACK_WIND (frame, pl_writev_cbk,
                            FIRST_CHILD (this), FIRST_CHILD (this)->fops->writev,
                            fd, vector, count, offset, iobref);

        if (op_ret == -1)
                STACK_UNWIND_STRICT (writev, frame, -1, op_errno, NULL, NULL);

        return 0;
}


int
pl_lk (call_frame_t *frame, xlator_t *this,
       fd_t *fd, int32_t cmd, struct flock *flock)
{
        void                  *transport = NULL;
        pid_t                  client_pid = 0;
        uint64_t               owner      = 0;
        posix_locks_private_t *priv = NULL;
        pl_inode_t            *pl_inode = NULL;
        int                    op_ret = 0;
        int                    op_errno = 0;
        int                    can_block = 0;
        posix_lock_t          *reqlock = NULL;
        posix_lock_t          *conf = NULL;
        int                    ret = 0;

        transport  = frame->root->trans;
        client_pid = frame->root->pid;
        owner      = frame->root->lk_owner;
        priv       = this->private;

        pl_inode = pl_inode_get (this, fd->inode);
        if (!pl_inode) {
                gf_log (this->name, GF_LOG_ERROR,
                        "Out of memory.");
                op_ret = -1;
                op_errno = ENOMEM;
                goto unwind;
        }

        reqlock = new_posix_lock (flock, transport, client_pid,
                                  owner, fd);

        if (!reqlock) {
                gf_log (this->name, GF_LOG_ERROR,
                        "Out of memory.");
                op_ret = -1;
                op_errno = ENOMEM;
                goto unwind;
        }

        pl_trace_in (this, frame, fd, NULL, cmd, flock, NULL);

        switch (cmd) {

#if F_GETLK != F_GETLK64
        case F_GETLK64:
#endif
        case F_GETLK:
                conf = pl_getlk (pl_inode, reqlock);
                posix_lock_to_flock (conf, flock);
                __destroy_lock (reqlock);

                break;

#if F_SETLKW != F_SETLKW64
        case F_SETLKW64:
#endif
        case F_SETLKW:
                can_block = 1;
                reqlock->frame  = frame;
                reqlock->this   = this;

                /* fall through */

#if F_SETLK != F_SETLK64
        case F_SETLK64:
#endif
        case F_SETLK:
                memcpy (&reqlock->user_flock, flock, sizeof (struct flock));
                ret = pl_setlk (this, pl_inode, reqlock,
                                can_block);

                if (ret == -1) {
                        if (can_block) {
                                pl_trace_block (this, frame, fd, NULL, cmd, flock, NULL);
                                goto out;
                        }
                        gf_log (this->name, GF_LOG_DEBUG, "returning EAGAIN");
                        op_ret = -1;
                        op_errno = EAGAIN;
                        __destroy_lock (reqlock);
                }
        }

unwind:
        pl_trace_out (this, frame, fd, NULL, cmd, flock, op_ret, op_errno, NULL);
        pl_update_refkeeper (this, fd->inode);
        STACK_UNWIND_STRICT (lk, frame, op_ret, op_errno, flock);
out:
        return 0;
}


/* TODO: this function just logs, no action required?? */
int
pl_forget (xlator_t *this,
           inode_t *inode)
{
        pl_inode_t   *pl_inode = NULL;

        posix_lock_t *ext_tmp = NULL;
        posix_lock_t *ext_l   = NULL;
	struct list_head posixlks_released;

        pl_inode_lock_t *ino_tmp = NULL;
        pl_inode_lock_t *ino_l   = NULL;
	struct list_head inodelks_released;

        pl_rw_req_t *rw_tmp = NULL;
        pl_rw_req_t *rw_req = NULL;

        pl_entry_lock_t *entry_tmp = NULL;
        pl_entry_lock_t *entry_l   = NULL;
	struct list_head entrylks_released;

        pl_dom_list_t *dom = NULL;
        pl_dom_list_t *dom_tmp = NULL;

	INIT_LIST_HEAD (&posixlks_released);
	INIT_LIST_HEAD (&inodelks_released);
	INIT_LIST_HEAD (&entrylks_released);

        pl_inode = pl_inode_get (this, inode);

	pthread_mutex_lock (&pl_inode->mutex);
	{

		if (!list_empty (&pl_inode->rw_list)) {
			gf_log (this->name, GF_LOG_WARNING,
				"Pending R/W requests found, releasing.");

			list_for_each_entry_safe (rw_req, rw_tmp, &pl_inode->rw_list,
						  list) {

				list_del (&rw_req->list);
				GF_FREE (rw_req);
			}
		}

		if (!list_empty (&pl_inode->ext_list)) {
			gf_log (this->name, GF_LOG_WARNING,
				"Pending fcntl locks found, releasing.");

			list_for_each_entry_safe (ext_l, ext_tmp, &pl_inode->ext_list,
						  list) {

				__delete_lock (pl_inode, ext_l);
				if (ext_l->blocked) {
					list_add_tail (&ext_l->list, &posixlks_released);
					continue;
				}
				__destroy_lock (ext_l);
			}
		}


		list_for_each_entry_safe (dom, dom_tmp, &pl_inode->dom_list, inode_list) {

			if (!list_empty (&dom->inodelk_list)) {
				gf_log (this->name, GF_LOG_WARNING,
					"Pending inode locks found, releasing.");

				list_for_each_entry_safe (ino_l, ino_tmp, &dom->inodelk_list, list) {
					__delete_inode_lock (ino_l);
					__destroy_inode_lock (ino_l);
				}

				list_splice_init (&dom->blocked_inodelks, &inodelks_released);
				

			}
			if (!list_empty (&dom->entrylk_list)) {
				gf_log (this->name, GF_LOG_WARNING,
					"Pending entry locks found, releasing.");

				list_for_each_entry_safe (entry_l, entry_tmp, &dom->entrylk_list, domain_list) {
					list_del_init (&entry_l->domain_list);

					if (entry_l->basename)
						GF_FREE ((char *)entry_l->basename);
					GF_FREE (entry_l);
				}

				list_splice_init (&dom->blocked_entrylks, &entrylks_released);
			}

			list_del (&dom->inode_list);
			gf_log ("posix-locks", GF_LOG_TRACE,
				" Cleaning up domain: %s", dom->domain);
			GF_FREE ((char *)(dom->domain));
			GF_FREE (dom);
		}

	}
	pthread_mutex_unlock (&pl_inode->mutex);

	list_for_each_entry_safe (ext_l, ext_tmp, &posixlks_released, list) {

		STACK_UNWIND_STRICT (lk, ext_l->frame, -1, 0, &ext_l->user_flock);
		__destroy_lock (ext_l);
	}

	list_for_each_entry_safe (ino_l, ino_tmp, &inodelks_released, blocked_locks) {

		STACK_UNWIND_STRICT (inodelk, ino_l->frame, -1, 0);
		__destroy_inode_lock (ino_l);
	}

	list_for_each_entry_safe (entry_l, entry_tmp, &entrylks_released, blocked_locks) {

		STACK_UNWIND_STRICT (entrylk, entry_l->frame, -1, 0);
		if (entry_l->basename)
			GF_FREE ((char *)entry_l->basename);
		GF_FREE (entry_l);

	}

        GF_FREE (pl_inode);

        return 0;
}

int
pl_release (xlator_t *this, fd_t *fd)
{
        pl_inode_t *pl_inode     = NULL;
        uint64_t    tmp_pl_inode = 0;
        int         ret          = -1;

	ret = inode_ctx_get (fd->inode, this, &tmp_pl_inode);
        if (ret != 0)
                goto out;

        pl_inode = (pl_inode_t *)(long)tmp_pl_inode;

        pl_trace_release (this, fd);

        gf_log (this->name, GF_LOG_TRACE,
                "Releasing all locks with fd %p", fd);

        delete_locks_of_fd (this, pl_inode, fd);

out:
        return ret;
}
static int32_t
__get_posixlk_count (xlator_t *this, pl_inode_t *pl_inode)
{
        posix_lock_t *lock   = NULL;
        int32_t       count  = 0;

        list_for_each_entry (lock, &pl_inode->ext_list, list) {

			gf_log (this->name, GF_LOG_DEBUG,
                                " XATTR DEBUG"
				"%s (pid=%d) (lk-owner=%"PRIu64") %"PRId64" - %"PRId64" state: %s",
				lock->fl_type == F_UNLCK ? "Unlock" : "Lock",
				lock->client_pid,
                                lock->owner,
				lock->user_flock.l_start,
				lock->user_flock.l_len,
                                lock->blocked == 1 ? "Blocked" : "Active");

                count++;
        }

        return count;
}

int32_t
get_posixlk_count (xlator_t *this, inode_t *inode)
{
        pl_inode_t   *pl_inode = NULL;
        uint64_t      tmp_pl_inode = 0;
        int           ret      = 0;
        int32_t       count    = 0;

        ret = inode_ctx_get (inode, this, &tmp_pl_inode);
        if (ret != 0) {
                goto out;
        }

        pl_inode = (pl_inode_t *)(long) tmp_pl_inode;

        pthread_mutex_lock (&pl_inode->mutex);
        {
                count =__get_posixlk_count (this, pl_inode);
        }
        pthread_mutex_unlock (&pl_inode->mutex);

out:
        return count;
}

void
pl_entrylk_xattr_fill (xlator_t *this, inode_t *inode,
                       dict_t *dict)
{
        int32_t     count = 0;
        int         ret   = -1;

        count = get_entrylk_count (this, inode);
        ret = dict_set_int32 (dict, GLUSTERFS_ENTRYLK_COUNT, count);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_DEBUG,
                        " dict_set failed on key %s", GLUSTERFS_ENTRYLK_COUNT);
        }

}

void
pl_inodelk_xattr_fill (xlator_t *this, inode_t *inode,
                       dict_t *dict)
{
        int32_t     count = 0;
        int         ret   = -1;

        count = get_inodelk_count (this, inode);
        ret = dict_set_int32 (dict, GLUSTERFS_INODELK_COUNT, count);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_DEBUG,
                        " dict_set failed on key %s", GLUSTERFS_INODELK_COUNT);
        }

}

void
pl_posixlk_xattr_fill (xlator_t *this, inode_t *inode,
                       dict_t *dict)
{
        int32_t     count = 0;
        int         ret   = -1;

        count = get_posixlk_count (this, inode);
        ret = dict_set_int32 (dict, GLUSTERFS_POSIXLK_COUNT, count);
        if (ret < 0) {
                gf_log (this->name, GF_LOG_DEBUG,
                        " dict_set failed on key %s", GLUSTERFS_POSIXLK_COUNT);
        }

}

int32_t
pl_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)
{
        pl_local_t *local = NULL;

        if (!frame->local) {
                goto out;
        }

        if (op_ret) {
                goto out;
        }

        local = frame->local;

        if (local->entrylk_count_req)
                pl_entrylk_xattr_fill (this, inode, dict);
        if (local->inodelk_count_req)
                pl_inodelk_xattr_fill (this, inode, dict);
        if (local->posixlk_count_req)
                pl_posixlk_xattr_fill (this, inode, dict);


        frame->local = NULL;

        if (local != NULL)
                GF_FREE (local);

out:
	STACK_UNWIND_STRICT (
                     lookup,
                     frame,
		      op_ret,
		      op_errno,
		      inode,
		      buf,
                      dict,
                      postparent);
{'fraction_non_alphanumeric': 0.09781327946869883, 'fraction_numerical': 0.007161547996143782, 'mean_word_length': 1.3981651376146789, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 453, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 74, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2537043', 'n_tokens_mistral': 22372, 'n_tokens_neox': 20296, 'n_words': 4636}