XrdFfsWcache.hh File Reference

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Functions
int	XrdFfsWcache_create (int fd, int flags)
void	XrdFfsWcache_destroy (int fd)
ssize_t	XrdFfsWcache_flush (int fd)
void	XrdFfsWcache_init (int basefd, int maxfd)
ssize_t	XrdFfsWcache_pread (int fd, char *buf, size_t len, off_t offset)
ssize_t	XrdFfsWcache_pwrite (int fd, char *buf, size_t len, off_t offset)

Function Documentation

XrdFfsWcache_create()

int XrdFfsWcache_create	(	int	fd,
		int	flags )

Definition at line 126 of file XrdFfsWcache.cc.

{
    XrdFfsWcache_destroy(fd);
    fd -= XrdFfsPosix_baseFD;
 
    XrdFfsWcacheFbufs[fd].offset = 0;
    XrdFfsWcacheFbufs[fd].len = 0;
    // "flag & O_RDONLY" is not equivalant to ! (flags & O_RDWR) && ! (flags & O_WRONLY)
    if ( ! (flags & O_RDWR) &&     
         ! (flags & O_WRONLY) &&
         (flags & O_DIRECT) )  // Limit the usage scenario of the read cache 
    {
        XrdFfsWcacheFbufs[fd].buf = (char*)malloc(XrdFfsRcacheBufsize);
        XrdFfsWcacheFbufs[fd].bufsize = XrdFfsRcacheBufsize;
    }
    else
    {
        XrdFfsWcacheFbufs[fd].buf = (char*)malloc(XrdFfsWcacheBufsize);
        XrdFfsWcacheFbufs[fd].bufsize = XrdFfsWcacheBufsize;
    }
    if (XrdFfsWcacheFbufs[fd].buf == NULL)
    {
        errno = ENOMEM;
        return 0;
    }
    XrdFfsWcacheFbufs[fd].mlock = (pthread_mutex_t*)malloc(sizeof(pthread_mutex_t));
    if (XrdFfsWcacheFbufs[fd].mlock == NULL)
    {
        errno = ENOMEM;
        return 0;
    }
    errno = pthread_mutex_init(XrdFfsWcacheFbufs[fd].mlock, NULL);
    if (errno)
        return 0;
    return 1;
}

References XrdFfsWcacheFilebuf::buf, XrdFfsWcacheFilebuf::mlock, O_DIRECT, XrdFfsPosix_baseFD, XrdFfsRcacheBufsize, XrdFfsWcache_destroy(), XrdFfsWcacheBufsize, and XrdFfsWcacheFbufs.

Here is the call graph for this function:

XrdFfsWcache_destroy()

void XrdFfsWcache_destroy

(

int

fd

)

Definition at line 170 of file XrdFfsWcache.cc.

{
/*  XrdFfsWcache_flush(fd); */
    fd -= XrdFfsPosix_baseFD;
 
    XrdFfsWcacheFbufs[fd].offset = 0;
    XrdFfsWcacheFbufs[fd].len = 0;
    if (XrdFfsWcacheFbufs[fd].buf != NULL) 
        free(XrdFfsWcacheFbufs[fd].buf);
    XrdFfsWcacheFbufs[fd].buf = NULL;
    if (XrdFfsWcacheFbufs[fd].mlock != NULL)
    {
        pthread_mutex_destroy(XrdFfsWcacheFbufs[fd].mlock);
        free(XrdFfsWcacheFbufs[fd].mlock);
    }
    XrdFfsWcacheFbufs[fd].mlock = NULL;
}

References XrdFfsWcacheFilebuf::buf, XrdFfsWcacheFilebuf::mlock, XrdFfsPosix_baseFD, and XrdFfsWcacheFbufs.

Referenced by XrdFfsWcache_create().

Here is the caller graph for this function:

XrdFfsWcache_flush()

ssize_t XrdFfsWcache_flush

(

int

fd

)

Definition at line 188 of file XrdFfsWcache.cc.

{
    ssize_t rc;
    fd -= XrdFfsPosix_baseFD;
 
    if (XrdFfsWcacheFbufs[fd].len == 0 || XrdFfsWcacheFbufs[fd].buf == NULL )
        return 0;
 
    rc = XrdFfsPosix_pwrite(fd + XrdFfsPosix_baseFD, 
                            XrdFfsWcacheFbufs[fd].buf, XrdFfsWcacheFbufs[fd].len, XrdFfsWcacheFbufs[fd].offset);
    if (rc > 0)
    {
        XrdFfsWcacheFbufs[fd].offset = 0;
        XrdFfsWcacheFbufs[fd].len = 0;
    }
    return rc;
}

References XrdFfsWcacheFilebuf::buf, XrdFfsWcacheFilebuf::len, XrdFfsWcacheFilebuf::offset, XrdFfsPosix_baseFD, XrdFfsPosix_pwrite(), and XrdFfsWcacheFbufs.

Referenced by XrdFfsWcache_pwrite().

Here is the call graph for this function:

Here is the caller graph for this function:

XrdFfsWcache_init()

void XrdFfsWcache_init	(	int	basefd,
		int	maxfd )

Definition at line 85 of file XrdFfsWcache.cc.

{
    int fd;
/* We are now using virtual file descriptors (from Xrootd Posix interface) in XrdFfsXrootdfs.cc so we need to set 
 * base (lowest) file descriptor, and max number of file descriptors..
 *
    struct rlimit rlp;
 
    getrlimit(RLIMIT_NOFILE, &rlp);
    XrdFfsWcacheNFILES = rlp.rlim_cur;
    XrdFfsWcacheNFILES = (XrdFfsWcacheNFILES == (int)RLIM_INFINITY? 4096 : XrdFfsWcacheNFILES);
 */
 
   XrdFfsPosix_baseFD = basefd;
   XrdFfsWcacheNFILES = maxfd; 
   
/*    printf("%d %d\n", XrdFfsWcacheNFILES, sizeof(struct XrdFfsWcacheFilebuf)); */
    XrdFfsWcacheFbufs = (struct XrdFfsWcacheFilebuf*)malloc(sizeof(struct XrdFfsWcacheFilebuf) * XrdFfsWcacheNFILES);
    for (fd = 0; fd < XrdFfsWcacheNFILES; fd++)
    {
        XrdFfsWcacheFbufs[fd].offset = 0;
        XrdFfsWcacheFbufs[fd].len = 0;
        XrdFfsWcacheFbufs[fd].buf = NULL;
        XrdFfsWcacheFbufs[fd].mlock = NULL;
    }
    if (!getenv("XRDCL_EC"))
    {
        XrdFfsRcacheBufsize = 1024 * 128;
    }
    else
    {
        char *savptr;
        int nbdat = atoi(strtok_r(getenv("XRDCL_EC"), ",", &savptr));
        strtok_r(NULL, ",", &savptr);
        int chsz = atoi(strtok_r(NULL, ",", &savptr));
        XrdFfsRcacheBufsize = nbdat * chsz; 
    }
    if (getenv("XROOTDFS_WCACHESZ"))
        XrdFfsRcacheBufsize = atoi(getenv("XROOTDFS_WCACHESZ"));
}

References XrdFfsPosix_baseFD, XrdFfsRcacheBufsize, XrdFfsWcacheFbufs, and XrdFfsWcacheNFILES.

XrdFfsWcache_pread()

ssize_t XrdFfsWcache_pread	(	int	fd,
		char *	buf,
		size_t	len,
		off_t	offset )

Definition at line 230 of file XrdFfsWcache.cc.

{
    ssize_t rc;
    fd -= XrdFfsPosix_baseFD;
    if (fd < 0)
    {
        errno = EBADF;
        return -1;
    }
 
    char *bufptr;
    size_t bufsize = XrdFfsWcacheFbufs[fd].bufsize;
 
    pthread_mutex_lock(XrdFfsWcacheFbufs[fd].mlock);
 
    // identity which block to cache
    if (XrdFfsWcacheFbufs[fd].len == 0 || 
        (offset / bufsize != XrdFfsWcacheFbufs[fd].offset / bufsize))
    {
        XrdFfsWcacheFbufs[fd].offset = (offset / bufsize) * bufsize;
        XrdFfsWcacheFbufs[fd].len = XrdFfsPosix_pread(fd + XrdFfsPosix_baseFD, 
                                                      XrdFfsWcacheFbufs[fd].buf,
                                                      bufsize,
                                                      XrdFfsWcacheFbufs[fd].offset);
    }  // when XrdFfsWcacheFbufs[fd].len < bufsize, the block is partially cached.
 
 
    // fetch data from the cache, up to the block's upper boundary.
    if (XrdFfsWcacheFbufs[fd].offset <= offset && 
        offset < XrdFfsWcacheFbufs[fd].offset + (off_t)XrdFfsWcacheFbufs[fd].len) 
    {  // read from cache, 
//----------------------------------------------------------
// FUSE doesn't like this block of the code, unless direct_io is enabled, or
// O_DIRECT flags is used. Otherwise, FUSES will stop reading prematurely
// when two processes read the same file at the same time.  
       bufptr = &XrdFfsWcacheFbufs[fd].buf[offset - XrdFfsWcacheFbufs[fd].offset]; 
       rc = (len < XrdFfsWcacheFbufs[fd].len - (offset - XrdFfsWcacheFbufs[fd].offset))?
             len : XrdFfsWcacheFbufs[fd].len - (offset - XrdFfsWcacheFbufs[fd].offset);
       memcpy(buf, bufptr, rc);
//----------------------------------------------------------
    } 
    else
    { // offset fall into the uncached part of the partically cached block
       rc = XrdFfsPosix_pread(fd + XrdFfsPosix_baseFD, buf, len, offset);
    }
    pthread_mutex_unlock(XrdFfsWcacheFbufs[fd].mlock);
/*    
    // prefetch the next block
    if ( (offset + rc) ==
         (XrdFfsWcacheFbufs[fd].offset + bufsize) )
    {
        pthread_t thread;
        pthread_attr_t attr;
        //size_t stacksize = 4*1024*1024;
    
        pthread_attr_init(&attr);
        //pthread_attr_setstacksize(&attr, stacksize);
        pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
 
        struct fd_n_offset nextblock(fd, (offset + bufsize));
        if (! pthread_create(&thread, &attr, XrdFfsWcache_updateReadCache, &nextblock)) 
            pthread_detach(thread);
        pthread_attr_destroy(&attr);
    }
*/
    return rc;
}

References XrdFfsWcacheFilebuf::buf, XrdFfsWcacheFilebuf::bufsize, XrdFfsWcacheFilebuf::len, XrdFfsWcacheFilebuf::mlock, XrdFfsWcacheFilebuf::offset, XrdFfsPosix_baseFD, XrdFfsPosix_pread(), and XrdFfsWcacheFbufs.

Here is the call graph for this function:

XrdFfsWcache_pwrite()

ssize_t XrdFfsWcache_pwrite	(	int	fd,
		char *	buf,
		size_t	len,
		off_t	offset )

Definition at line 298 of file XrdFfsWcache.cc.

{
    ssize_t rc;
    char *bufptr;
    fd -= XrdFfsPosix_baseFD;
    if (fd < 0)
    {
        errno = EBADF;
        return -1;
    }
 
/* do not use caching under these cases */
    if (len > (size_t)(XrdFfsWcacheBufsize/2) || fd >= XrdFfsWcacheNFILES)
    {
        rc = XrdFfsPosix_pwrite(fd + XrdFfsPosix_baseFD, buf, len, offset);
        return rc;
    }
 
    pthread_mutex_lock(XrdFfsWcacheFbufs[fd].mlock);
    rc = XrdFfsWcacheFbufs[fd].len;
/* 
   in the following two cases, a XrdFfsWcache_flush is required:
   1. current offset isnn't pointing to the tail of data in buffer
   2. adding new data will exceed the current buffer 
*/ 
    if (offset != (off_t)(XrdFfsWcacheFbufs[fd].offset + XrdFfsWcacheFbufs[fd].len) ||
        (off_t)(offset + len) > (XrdFfsWcacheFbufs[fd].offset + XrdFfsWcacheBufsize))
        rc = XrdFfsWcache_flush(fd + XrdFfsPosix_baseFD);
 
    errno = 0;
    if (rc < 0) 
    {
        errno = ENOSPC;
        pthread_mutex_unlock(XrdFfsWcacheFbufs[fd].mlock);
        return -1;
    }
 
    bufptr = &XrdFfsWcacheFbufs[fd].buf[XrdFfsWcacheFbufs[fd].len];
    memcpy(bufptr, buf, len);
    if (XrdFfsWcacheFbufs[fd].len == 0)
        XrdFfsWcacheFbufs[fd].offset = offset;
    XrdFfsWcacheFbufs[fd].len += len;
 
    pthread_mutex_unlock(XrdFfsWcacheFbufs[fd].mlock);
    return (ssize_t)len;
}

References XrdFfsWcacheFilebuf::buf, XrdFfsWcacheFilebuf::len, XrdFfsWcacheFilebuf::mlock, XrdFfsWcacheFilebuf::offset, XrdFfsPosix_baseFD, XrdFfsPosix_pwrite(), XrdFfsWcache_flush(), XrdFfsWcacheBufsize, XrdFfsWcacheFbufs, and XrdFfsWcacheNFILES.

Here is the call graph for this function: