/*
* Library for Posix async read operations with hints.
* Author: Don Capps
* Company: Iozone
* Date: 4/24/1998
*
* Two models are supported. First model is a replacement for read() where the async
* operations are performed and the requested data is bcopy()-ed back into the users
* buffer. The second model is a new version of read() where the caller does not
* supply the address of the buffer but instead is returned an address to the
* location of the data. The second model eliminates a bcopy from the path.
*
* To use model #1:
* 1. Call async_init(&pointer_on_stack,fd,direct_flag);
* The fd is the file descriptor for the async operations.
* The direct_flag sets VX_DIRECT
*
* 2. Call async_read(gc, fd, ubuffer, offset, size, stride, max, depth)
* Where:
* gc ............ is the pointer on the stack
* fd ............ is the file descriptor
* ubuffer ....... is the address of the user buffer.
* offset ........ is the offset in the file to begin reading
* size .......... is the size of the transfer.
* stride ........ is the distance, in size units, to space the async reads.
* max ........... is the max size of the file to be read.
* depth ......... is the number of async operations to perform.
*
* 3. Call end_async(gc) when finished.
* Where:
* gc ............ is the pointer on the stack.
*
* To use model #2:
* 1. Call async_init(&pointer_on_stack,fd,direct_flag);
* The fd is the file descriptor for the async operations.
* The direct_flag sets VX_DIRECT
* 2. Call async_read(gc, fd, &ubuffer, offset, size, stride, max, depth)
* Where:
* gc ............ is the pointer on the stack
* fd ............ is the file descriptor
* ubuffer ....... is the address of a pointer that will be filled in
* by the async library.
* offset ........ is the offset in the file to begin reading
* size .......... is the size of the transfer.
* stride ........ is the distance, in size units, to space the async reads.
* max ........... is the max size of the file to be read.
* depth ......... is the number of async operations to perform.
*
* 3. Call async_release(gc) when finished with the data that was returned.
* This allows the async library to reuse the memory that was filled in
* and returned to the user.
*
* 4. Call end_async(gc) when finished.
* Where:
* gc ............ is the pointer on the stack.
*
* To use model #1: (WRITES)
* 1. Call async_init(&pointer_on_stack,fd,direct_flag);
* The fd is the file descriptor for the async operations.
*
* 2. Call async_write(gc, fd, ubuffer, size, offset, depth)
* Where:
* gc ............ is the pointer on the stack
* fd ............ is the file descriptor
* ubuffer ....... is the address of the user buffer.
* size .......... is the size of the transfer.
* offset ........ is the offset in the file to begin reading
* depth ......... is the number of async operations to perform.
*
* 4. Call end_async(gc) when finished.
* Where:
* gc ............ is the pointer on the stack.
*
* Notes:
* The intended use is to replace calls to read() with calls to
* async_read() and allow the user to make suggestions on
* what kind of async read-ahead would be nice to have.
* The first transfer requested is guarenteed to be complete
* before returning to the caller. The async operations will
* be started and will also be guarenteed to have completed
* if the next call specifies its first request to be one
* that was previously performed with an async operation.
*
* The async_read_no_copy() function allows the async operations
* to return the data to the user and not have to perform
* a bcopy of the data back into the user specified buffer
* location. This model is faster but assumes that the user
* application has been modified to work with this model.
*
* The async_write() is intended to enhance the performance of
* initial writes to a file. This is the slowest case in the write
* path as it must perform meta-data allocations and wait.
*/
#include <sys/types.h>
#include <aio.h>
#if defined(solaris) || defined(linux) || defined(SCO_Unixware_gcc)
#else
#include <sys/timers.h>
#endif
#include <sys/errno.h>
#include <unistd.h>
#ifndef bsd4_4
#include <malloc.h>
#endif
#ifdef VXFS
#include <sys/fs/vx_ioctl.h>
#endif
#if defined(OSFV5) || defined(linux)
#include <string.h>
#endif
#if defined(linux)
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#endif
#if (defined(solaris) && defined(__LP64__)) || defined(__s390x__) || defined(FreeBSD)
/* If we are building for 64-bit Solaris, all functions that return pointers
* must be declared before they are used; otherwise the compiler will assume
* that they return ints and the top 32 bits of the pointer will be lost,
* causing segmentation faults. The following includes take care of this.
* It should be safe to add these for all other OSs too, but we're only
* doing it for Solaris now in case another OS turns out to be a special case.
*/
#include <stdio.h>
#include <stdlib.h>
#include <strings.h> /* For the BSD string functions */
#endif
void mbcopy(char *source, char *dest, size_t len);
#if !defined(solaris) && !defined(off64_t) && !defined(_OFF64_T) && !defined(__off64_t_defined) && !defined(SCO_Unixware_gcc)
typedef long long off64_t;
#endif
#if defined(OSFV5)
#include <string.h>
#endif
extern long long page_size;
extern int one;
/*
* Internal cache entrys. Each entry on the global
* cache, pointed to by async_init(gc) will be of
* this structure type.
*/
char version[] = "Libasync Version $Revision$";
struct cache_ent {
struct aiocb myaiocb; /* For use in small file mode */
#ifdef _LARGEFILE64_SOURCE
#if defined(__CrayX1__)
aiocb64_t myaiocb64; /* For use in large file mode */
#else
struct aiocb64 myaiocb64; /* For use in large file mode */
#endif
#endif
long long fd; /* File descriptor */
long long size; /* Size of the transfer */
struct cache_ent *forward; /* link to next element on cache list */
struct cache_ent *back; /* link to previous element on the cache list */
long long direct; /* flag to indicate if the buffer should be */
/* de-allocated by library */
char *real_address; /* Real address to free */
volatile void *oldbuf; /* Used for firewall to prevent in flight */
/* accidents */
int oldfd; /* Used for firewall to prevent in flight */
/* accidents */
size_t oldsize; /* Used for firewall to prevent in flight */
/* accidents */
};
/*
* Head of the cache list
*/
struct cache {
struct cache_ent *head; /* Head of cache list */
struct cache_ent *tail; /* tail of cache list */
struct cache_ent *inuse_head; /* head of in-use list */
long long count; /* How many elements on the cache list */
struct cache_ent *w_head; /* Head of cache list */
struct cache_ent *w_tail; /* tail of cache list */
long long w_count; /* How many elements on the write list */
};
long long max_depth;
extern int errno;
struct cache_ent *alloc_cache();
struct cache_ent *incache();
void async_init();
void end_async();
int async_suspend();
int async_read();
void takeoff_cache();
void del_cache();
void async_release();
void putoninuse();
void takeoffinuse();
struct cache_ent *allocate_write_buffer();
size_t async_write();
void async_wait_for_write();
void async_put_on_write_queue();
void async_write_finish();
/* On Solaris _LP64 will be defined by <sys/types.h> if we're compiling
* as a 64-bit binary. Make sure that __LP64__ gets defined in this case,
* too -- it should be defined on the compiler command line, but let's
* not rely on this.
*/
#if defined(_LP64)
#if !defined(__LP64__)
#define __LP64__
#endif
#endif
/***********************************************/
/* Initialization routine to setup the library */
/***********************************************/
void
async_init(gc,fd,flag)
struct cache **gc;
int fd;
int flag;
{
#ifdef VXFS
if(flag)
ioctl(fd,VX_SETCACHE,VX_DIRECT);
#endif
if(*gc)
{
printf("Warning calling async_init two times ?\n");
return;
}
*gc=(struct cache *)malloc((size_t)sizeof(struct cache));
if(*gc == 0)
{
printf("Malloc failed\n");
exit(174);
}
bzero(*gc,sizeof(struct cache));
#if defined(__AIX__) || defined(SCO_Unixware_gcc)
max_depth=500;
#else
max_depth=sysconf(_SC_AIO_MAX);
#endif
}
/***********************************************/
/* Tear down routine to shutdown the library */
/***********************************************/
void
end_async(gc)
struct cache *gc;
{
del_cache(gc);
async_write_finish(gc);
free((void *)gc);
}
/***********************************************/
/* Wait for a request to finish */
/***********************************************/
int
async_suspend(struct cache_ent *ce)
{
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
const struct aiocb * const cblist[1] = {&ce->myaiocb};
#else
const struct aiocb64 * const cblist[1] = {&ce->myaiocb64};
#endif
#else
const struct aiocb * const cblist[1] = {&ce->myaiocb};
#endif
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
return aio_suspend(cblist, 1, NULL);
#else
return aio_suspend64(cblist, 1, NULL);
#endif
#else
return aio_suspend(cblist, 1, NULL);
#endif
}
/*************************************************************************
* This routine is a generic async reader assist funtion. It takes
* the same calling parameters as read() but also extends the
* interface to include:
* stride ..... For the async reads, what is the distance, in size units,
* to space the reads. Note: Stride of 0 indicates that
* you do not want any read-ahead.
* max ..... What is the maximum file offset for this operation.
* depth ..... How much read-ahead do you want.
*
* The calls to this will guarentee to complete the read() operation
* before returning to the caller. The completion may occur in two
* ways. First the operation may be completed by calling aio_read()
* and then waiting for it to complete. Second the operation may be
* completed by copying the data from a cache of previously completed
* async operations.
* In the event the read to be satisfied is not in the cache then a
* series of async operations will be scheduled and then the first
* async read will be completed. In the event that the read() can be
* satisfied from the cache then the data is copied back to the
* user buffer and a series of async reads will be initiated. If a
* read is issued and the cache contains data and the read can not
* be satisfied from the cache, then the cache is discarded, and
* a new cache is constructed.
* Note: All operations are aio_read(). The series will be issued
* as asyncs in the order requested. After all are in flight
* then the code will wait for the manditory first read.
*************************************************************************/
int
async_read(gc, fd, ubuffer, offset, size, stride, max, depth)
struct cache *gc;
long long fd;
char *ubuffer;
off64_t offset;
long long size;
long long stride;
off64_t max;
long long depth;
{
off64_t a_offset,r_offset;
long long a_size;
struct cache_ent *ce,*first_ce=0;
long long i;
ssize_t retval=0;
ssize_t ret;
long long start = 0;
long long del_read=0;
a_offset=offset;
a_size = size;
/*
* Check to see if it can be completed from the cache
*/
if((ce=(struct cache_ent *)incache(gc,fd,offset,size)))
{
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
while((ret=aio_error(&ce->myaiocb))== EINPROGRESS)
{
async_suspend(ce);
}
#else
while((ret=aio_error64(&ce->myaiocb64))== EINPROGRESS)
{
async_suspend(ce);
}
#endif
#else
while((ret=aio_error(&ce->myaiocb))== EINPROGRESS)
{
async_suspend(ce);
}
#endif
if(ret)
{
printf("aio_error 1: ret %d %d\n",ret,errno);
}
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
retval=aio_return(&ce->myaiocb);
#else
#if defined(__CrayX1__)
retval=aio_return64((aiocb64_t *)&ce->myaiocb64);
#else
retval=aio_return64((struct aiocb64 *)&ce->myaiocb64);
#endif
#endif
#else
retval=aio_return(&ce->myaiocb);
#endif
if(retval > 0)
{
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
mbcopy((char *)ce->myaiocb.aio_buf,(char *)ubuffer,(size_t)retval);
#else
mbcopy((char *)ce->myaiocb64.aio_buf,(char *)ubuffer,(size_t)retval);
#endif
#else
mbcopy((char *)ce->myaiocb.aio_buf,(char *)ubuffer,(size_t)retval);
#endif
}
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
if(retval < ce->myaiocb.aio_nbytes)
#else
if(retval < ce->myaiocb64.aio_nbytes)
#endif
#else
if(retval < ce->myaiocb.aio_nbytes)
#endif
{
printf("aio_return error1: ret %d %d\n",retval,errno);
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
printf("aio_return error1: fd %d offset %ld buffer %lx size %d Opcode %d\n",
ce->myaiocb.aio_fildes,
ce->myaiocb.aio_offset,
(long)(ce->myaiocb.aio_buf),
ce->myaiocb.aio_nbytes,
ce->myaiocb.aio_lio_opcode
#else
printf("aio_return error1: fd %d offset %lld buffer %lx size %d Opcode %d\n",
ce->myaiocb64.aio_fildes,
ce->myaiocb64.aio_offset,
(long)(ce->myaiocb64.aio_buf),
ce->myaiocb64.aio_nbytes,
ce->myaiocb64.aio_lio_opcode
#endif
#else
printf("aio_return error1: fd %d offset %d buffer %lx size %d Opcode %d\n",
ce->myaiocb.aio_fildes,
ce->myaiocb.aio_offset,
(long)(ce->myaiocb.aio_buf),
ce->myaiocb.aio_nbytes,
ce->myaiocb.aio_lio_opcode
#endif
);
}
ce->direct=0;
takeoff_cache(gc,ce);
}else
{
/*
* Clear the cache and issue the first request async()
*/
del_cache(gc);
del_read++;
first_ce=alloc_cache(gc,fd,offset,size,(long long)LIO_READ);
again:
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
ret=aio_read(&first_ce->myaiocb);
#else
ret=aio_read64(&first_ce->myaiocb64);
#endif
#else
ret=aio_read(&first_ce->myaiocb);
#endif
if(ret!=0)
{
if(errno==EAGAIN)
goto again;
else
printf("error returned from aio_read(). Ret %d errno %d\n",ret,errno);
}
}
if(stride==0) /* User does not want read-ahead */
goto out;
if(a_offset<0) /* Before beginning of file */
goto out;
if(a_offset+size>max) /* After end of file */
goto out;
if(depth >=(max_depth-1))
depth=max_depth-1;
if(depth==0)
goto out;
if(gc->count > 1)
start=depth-1;
for(i=start;i<depth;i++) /* Issue read-aheads for the depth specified */
{
r_offset=a_offset+((i+1)*(stride*a_size));
if(r_offset<0)
continue;
if(r_offset+size > max)
continue;
if((ce=incache(gc,fd,r_offset,a_size)))
continue;
ce=alloc_cache(gc,fd,r_offset,a_size,(long long)LIO_READ);
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
ret=aio_read(&ce->myaiocb);
#else
ret=aio_read64(&ce->myaiocb64);
#endif
#else
ret=aio_read(&ce->myaiocb);
#endif
if(ret!=0)
{
takeoff_cache(gc,ce);
break;
}
}
out:
if(del_read) /* Wait for the first read to complete */
{
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
while((ret=aio_error(&first_ce->myaiocb))== EINPROGRESS)
{
async_suspend(first_ce);
}
#else
while((ret=aio_error64(&first_ce->myaiocb64))== EINPROGRESS)
{
async_suspend(first_ce);
}
#endif
#else
while((ret=aio_error(&first_ce->myaiocb))== EINPROGRESS)
{
async_suspend(first_ce);
}
#endif
if(ret)
printf("aio_error 2: ret %d %d\n",ret,errno);
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
retval=aio_return(&first_ce->myaiocb);
#else
retval=aio_return64(&first_ce->myaiocb64);
#endif
#else
retval=aio_return(&first_ce->myaiocb);
#endif
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
if(retval < first_ce->myaiocb.aio_nbytes)
#else
if(retval < first_ce->myaiocb64.aio_nbytes)
#endif
#else
if(retval < first_ce->myaiocb.aio_nbytes)
#endif
{
printf("aio_return error2: ret %d %d\n",retval,errno);
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
printf("aio_return error2: fd %d offset %lld buffer %lx size %d Opcode %d\n",
first_ce->myaiocb.aio_fildes,
first_ce->myaiocb.aio_offset,
(long)(first_ce->myaiocb.aio_buf),
first_ce->myaiocb.aio_nbytes,
first_ce->myaiocb.aio_lio_opcode
#else
printf("aio_return error2: fd %d offset %lld buffer %lx size %d Opcode %d\n",
first_ce->myaiocb64.aio_fildes,
first_ce->myaiocb64.aio_offset,
(long)(first_ce->myaiocb64.aio_buf),
first_ce->myaiocb64.aio_nbytes,
first_ce->myaiocb64.aio_lio_opcode
#endif
#else
printf("aio_return error2: fd %d offset %d buffer %lx size %d Opcode %d\n",
first_ce->myaiocb.aio_fildes,
first_ce->myaiocb.aio_offset,
(long)(first_ce->myaiocb.aio_buf),
first_ce->myaiocb.aio_nbytes,
first_ce->myaiocb.aio_lio_opcode
#endif
);
}
if(retval > 0)
{
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
mbcopy((char *)first_ce->myaiocb.aio_buf,(char *)ubuffer,(size_t)retval);
#else
mbcopy((char *)first_ce->myaiocb64.aio_buf,(char *)ubuffer,(size_t)retval);
#endif
#else
mbcopy((char *)first_ce->myaiocb.aio_buf,(char *)ubuffer,(size_t)retval);
#endif
}
first_ce->direct=0;
takeoff_cache(gc,first_ce);
}
return((int)retval);
}
/************************************************************************
* This routine allocates a cache_entry. It contains the
* aiocb block as well as linkage for use in the cache mechanism.
* The space allocated here will be released after the cache entry
* has been consumed. The routine takeoff_cache() will be called
* after the data has been copied to user buffer or when the
* cache is purged. The routine takeoff_cache() will also release
* all memory associated with this cache entry.
************************************************************************/
struct cache_ent *
alloc_cache(gc,fd,offset,size,op)
struct cache *gc;
long long fd,size,op;
off64_t offset;
{
struct cache_ent *ce;
long temp;
ce=(struct cache_ent *)malloc((size_t)sizeof(struct cache_ent));
if(ce == (struct cache_ent *)0)
{
printf("Malloc failed\n");
exit(175);
}
bzero(ce,sizeof(struct cache_ent));
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
ce->myaiocb.aio_fildes=(int)fd;
ce->myaiocb.aio_offset=(off64_t)offset;
ce->real_address = (char *)malloc((size_t)(size+page_size));
temp=(long)ce->real_address;
temp = (temp+page_size) & ~(page_size-1);
ce->myaiocb.aio_buf=(volatile void *)temp;
if(ce->myaiocb.aio_buf == 0)
#else
ce->myaiocb64.aio_fildes=(int)fd;
ce->myaiocb64.aio_offset=(off64_t)offset;
ce->real_address = (char *)malloc((size_t)(size+page_size));
temp=(long)ce->real_address;
temp = (temp+page_size) & ~(page_size-1);
ce->myaiocb64.aio_buf=(volatile void *)temp;
if(ce->myaiocb64.aio_buf == 0)
#endif
#else
ce->myaiocb.aio_fildes=(int)fd;
ce->myaiocb.aio_offset=(off_t)offset;
ce->real_address = (char *)malloc((size_t)(size+page_size));
temp=(long)ce->real_address;
temp = (temp+page_size) & ~(page_size-1);
ce->myaiocb.aio_buf=(volatile void *)temp;
if(ce->myaiocb.aio_buf == 0)
#endif
{
printf("Malloc failed\n");
exit(176);
}
/*bzero(ce->myaiocb.aio_buf,(size_t)size);*/
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
ce->myaiocb.aio_reqprio=0;
ce->myaiocb.aio_nbytes=(size_t)size;
ce->myaiocb.aio_sigevent.sigev_notify=SIGEV_NONE;
ce->myaiocb.aio_lio_opcode=(int)op;
#else
ce->myaiocb64.aio_reqprio=0;
ce->myaiocb64.aio_nbytes=(size_t)size;
ce->myaiocb64.aio_sigevent.sigev_notify=SIGEV_NONE;
ce->myaiocb64.aio_lio_opcode=(int)op;
#endif
#else
ce->myaiocb.aio_reqprio=0;
ce->myaiocb.aio_nbytes=(size_t)size;
ce->myaiocb.aio_sigevent.sigev_notify=SIGEV_NONE;
ce->myaiocb.aio_lio_opcode=(int)op;
#endif
ce->fd=(int)fd;
ce->forward=0;
ce->back=gc->tail;
if(gc->tail)
gc->tail->forward = ce;
gc->tail= ce;
if(!gc->head)
gc->head=ce;
gc->count++;
return(ce);
}
/************************************************************************
* This routine checks to see if the requested data is in the
* cache.
*************************************************************************/
struct cache_ent *
incache(gc,fd,offset,size)
struct cache *gc;
long long fd,size;
off64_t offset;
{
struct cache_ent *move;
if(gc->head==0)
{
return(0);
}
move=gc->head;
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
while(move)
{
if((move->fd == fd) && (move->myaiocb.aio_offset==(off64_t)offset) &&
((size_t)size==move->myaiocb.aio_nbytes))
{
return(move);
}
move=move->forward;
}
#else
while(move)
{
if((move->fd == fd) && (move->myaiocb64.aio_offset==(off64_t)offset) &&
((size_t)size==move->myaiocb64.aio_nbytes))
{
return(move);
}
move=move->forward;
}
#endif
#else
while(move)
{
if((move->fd == fd) && (move->myaiocb.aio_offset==(off_t)offset) &&
((size_t)size==move->myaiocb.aio_nbytes))
{
return(move);
}
move=move->forward;
}
#endif
return(0);
}
/************************************************************************
* This routine removes a specific cache entry from the cache, and
* releases all memory associated witht the cache entry (if not direct).
*************************************************************************/
void
takeoff_cache(gc,ce)
struct cache *gc;
struct cache_ent *ce;
{
struct cache_ent *move;
long long found;
move=gc->head;
if(move==ce) /* Head of list */
{
gc->head=ce->forward;
if(gc->head)
gc->head->back=0;
else
gc->tail = 0;
if(!ce->direct)
{
free((void *)(ce->real_address));
free((void *)ce);
}
gc->count--;
return;
}
found=0;
while(move)
{
if(move==ce)
{
if(move->forward)
{
move->forward->back=move->back;
}
if(move->back)
{
move->back->forward=move->forward;
}
found=1;
break;
}
else
{
move=move->forward;
}
}
if(gc->head == ce)
gc->tail = ce;
if(!found)
printf("Internal Error in takeoff cache\n");
move=gc->head;
if(!ce->direct)
{
free((void *)(ce->real_address));
free((void *)ce);
}
gc->count--;
}
/************************************************************************
* This routine is used to purge the entire cache. This is called when
* the cache contains data but the incomming read was not able to
* be satisfied from the cache. This indicates that the previous
* async read-ahead was not correct and a new pattern is emerging.
************************************************************************/
void
del_cache(gc)
struct cache *gc;
{
struct cache_ent *ce;
ssize_t ret;
ce=gc->head;
while(1)
{
ce=gc->head;
if(ce==0)
return;
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
while((ret = aio_cancel(0,&ce->myaiocb))==AIO_NOTCANCELED)
#else
while((ret = aio_cancel64(0,&ce->myaiocb64))==AIO_NOTCANCELED)
#endif
#else
while((ret = aio_cancel(0,&ce->myaiocb))==AIO_NOTCANCELED)
#endif
;
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
ret = aio_return(&ce->myaiocb);
#else
ret = aio_return64(&ce->myaiocb64);
#endif
#else
ret = aio_return(&ce->myaiocb);
#endif
ce->direct=0;
takeoff_cache(gc,ce); /* remove from cache */
}
}
/************************************************************************
* Like its sister async_read() this function performs async I/O for
* all buffers but it differs in that it expects the caller to
* request a pointer to the data to be returned instead of handing
* the function a location to put the data. This will allow the
* async I/O to be performed and does not require any bcopy to be
* done to put the data back into the location specified by the caller.
************************************************************************/
int
async_read_no_copy(gc, fd, ubuffer, offset, size, stride, max, depth)
struct cache *gc;
long long fd;
char **ubuffer;
off64_t offset;
long long size;
long long stride;
off64_t max;
long long depth;
{
off64_t a_offset,r_offset;
long long a_size;
struct cache_ent *ce,*first_ce=0;
long long i;
ssize_t retval=0;
ssize_t ret;
long long del_read=0;
long long start=0;
a_offset=offset;
a_size = size;
/*
* Check to see if it can be completed from the cache
*/
if((ce=(struct cache_ent *)incache(gc,fd,offset,size)))
{
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
while((ret=aio_error(&ce->myaiocb))== EINPROGRESS)
{
async_suspend(ce);
}
#else
while((ret=aio_error64(&ce->myaiocb64))== EINPROGRESS)
{
async_suspend(ce);
}
#endif
#else
while((ret=aio_error(&ce->myaiocb))== EINPROGRESS)
{
async_suspend(ce);
}
#endif
if(ret)
printf("aio_error 3: ret %d %d\n",ret,errno);
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
if(ce->oldbuf != ce->myaiocb.aio_buf ||
ce->oldfd != ce->myaiocb.aio_fildes ||
ce->oldsize != ce->myaiocb.aio_nbytes)
#else
if(ce->oldbuf != ce->myaiocb64.aio_buf ||
ce->oldfd != ce->myaiocb64.aio_fildes ||
ce->oldsize != ce->myaiocb64.aio_nbytes)
#endif
#else
if(ce->oldbuf != ce->myaiocb.aio_buf ||
ce->oldfd != ce->myaiocb.aio_fildes ||
ce->oldsize != ce->myaiocb.aio_nbytes)
#endif
printf("It changed in flight\n");
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
retval=aio_return(&ce->myaiocb);
#else
retval=aio_return64(&ce->myaiocb64);
#endif
#else
retval=aio_return(&ce->myaiocb);
#endif
if(retval > 0)
{
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
*ubuffer=(char *)ce->myaiocb.aio_buf;
#else
*ubuffer=(char *)ce->myaiocb64.aio_buf;
#endif
#else
*ubuffer=(char *)ce->myaiocb.aio_buf;
#endif
}else
*ubuffer=0;
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
if(retval < ce->myaiocb.aio_nbytes)
#else
if(retval < ce->myaiocb64.aio_nbytes)
#endif
#else
if(retval < ce->myaiocb.aio_nbytes)
#endif
{
printf("aio_return error4: ret %d %d\n",retval,errno);
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
printf("aio_return error4: fd %d offset %lld buffer %lx size %d Opcode %d\n",
ce->myaiocb.aio_fildes,
ce->myaiocb.aio_offset,
(long)(ce->myaiocb.aio_buf),
ce->myaiocb.aio_nbytes,
ce->myaiocb.aio_lio_opcode
#else
printf("aio_return error4: fd %d offset %lld buffer %lx size %d Opcode %d\n",
ce->myaiocb64.aio_fildes,
ce->myaiocb64.aio_offset,
(long)(ce->myaiocb64.aio_buf),
ce->myaiocb64.aio_nbytes,
ce->myaiocb64.aio_lio_opcode
#endif
#else
printf("aio_return error4: fd %d offset %d buffer %lx size %d Opcode %d\n",
ce->myaiocb.aio_fildes,
ce->myaiocb.aio_offset,
(long)(ce->myaiocb.aio_buf),
ce->myaiocb.aio_nbytes,
ce->myaiocb.aio_lio_opcode
#endif
);
}
ce->direct=1;
takeoff_cache(gc,ce); /* do not delete buffer*/
putoninuse(gc,ce);
}else
{
/*
* Clear the cache and issue the first request async()
*/
del_cache(gc);
del_read++;
first_ce=alloc_cache(gc,fd,offset,size,(long long)LIO_READ); /* allocate buffer */
/*printf("allocated buffer/read %x offset %d\n",first_ce->myaiocb.aio_buf,offset);*/
again:
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
first_ce->oldbuf=first_ce->myaiocb.aio_buf;
first_ce->oldfd=first_ce->myaiocb.aio_fildes;
first_ce->oldsize=first_ce->myaiocb.aio_nbytes;
ret=aio_read(&first_ce->myaiocb);
#else
first_ce->oldbuf=first_ce->myaiocb64.aio_buf;
first_ce->oldfd=first_ce->myaiocb64.aio_fildes;
first_ce->oldsize=first_ce->myaiocb64.aio_nbytes;
ret=aio_read64(&first_ce->myaiocb64);
#endif
#else
first_ce->oldbuf=first_ce->myaiocb.aio_buf;
first_ce->oldfd=first_ce->myaiocb.aio_fildes;
first_ce->oldsize=first_ce->myaiocb.aio_nbytes;
ret=aio_read(&first_ce->myaiocb);
#endif
if(ret!=0)
{
if(errno==EAGAIN)
goto again;
else
printf("error returned from aio_read(). Ret %d errno %d\n",ret,errno);
}
}
if(stride==0) /* User does not want read-ahead */
goto out;
if(a_offset<0) /* Before beginning of file */
goto out;
if(a_offset+size>max) /* After end of file */
goto out;
if(depth >=(max_depth-1))
depth=max_depth-1;
if(depth==0)
goto out;
if(gc->count > 1)
start=depth-1;
for(i=start;i<depth;i++) /* Issue read-aheads for the depth specified */
{
r_offset=a_offset+((i+1)*(stride*a_size));
if(r_offset<0)
continue;
if(r_offset+size > max)
continue;
if((ce=incache(gc,fd,r_offset,a_size)))
continue;
ce=alloc_cache(gc,fd,r_offset,a_size,(long long)LIO_READ);
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
ce->oldbuf=ce->myaiocb.aio_buf;
ce->oldfd=ce->myaiocb.aio_fildes;
ce->oldsize=ce->myaiocb.aio_nbytes;
ret=aio_read(&ce->myaiocb);
#else
ce->oldbuf=ce->myaiocb64.aio_buf;
ce->oldfd=ce->myaiocb64.aio_fildes;
ce->oldsize=ce->myaiocb64.aio_nbytes;
ret=aio_read64(&ce->myaiocb64);
#endif
#else
ce->oldbuf=ce->myaiocb.aio_buf;
ce->oldfd=ce->myaiocb.aio_fildes;
ce->oldsize=ce->myaiocb.aio_nbytes;
ret=aio_read(&ce->myaiocb);
#endif
if(ret!=0)
{
takeoff_cache(gc,ce);
break;
}
}
out:
if(del_read) /* Wait for the first read to complete */
{
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
while((ret=aio_error(&first_ce->myaiocb))== EINPROGRESS)
{
async_suspend(first_ce);
}
#else
while((ret=aio_error64(&first_ce->myaiocb64))== EINPROGRESS)
{
async_suspend(first_ce);
}
#endif
#else
while((ret=aio_error(&first_ce->myaiocb))== EINPROGRESS)
{
async_suspend(first_ce);
}
#endif
if(ret)
printf("aio_error 4: ret %d %d\n",ret,errno);
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
if(first_ce->oldbuf != first_ce->myaiocb.aio_buf ||
first_ce->oldfd != first_ce->myaiocb.aio_fildes ||
first_ce->oldsize != first_ce->myaiocb.aio_nbytes)
printf("It changed in flight2\n");
retval=aio_return(&first_ce->myaiocb);
#else
if(first_ce->oldbuf != first_ce->myaiocb64.aio_buf ||
first_ce->oldfd != first_ce->myaiocb64.aio_fildes ||
first_ce->oldsize != first_ce->myaiocb64.aio_nbytes)
printf("It changed in flight2\n");
retval=aio_return64(&first_ce->myaiocb64);
#endif
#else
if(first_ce->oldbuf != first_ce->myaiocb.aio_buf ||
first_ce->oldfd != first_ce->myaiocb.aio_fildes ||
first_ce->oldsize != first_ce->myaiocb.aio_nbytes)
printf("It changed in flight2\n");
retval=aio_return(&first_ce->myaiocb);
#endif
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
if(retval < first_ce->myaiocb.aio_nbytes)
#else
if(retval < first_ce->myaiocb64.aio_nbytes)
#endif
#else
if(retval < first_ce->myaiocb.aio_nbytes)
#endif
{
printf("aio_return error5: ret %d %d\n",retval,errno);
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
printf("aio_return error5: fd %d offset %lld buffer %lx size %d Opcode %d\n",
first_ce->myaiocb.aio_fildes,
first_ce->myaiocb.aio_offset,
(long)(first_ce->myaiocb.aio_buf),
first_ce->myaiocb.aio_nbytes,
first_ce->myaiocb.aio_lio_opcode
#else
printf("aio_return error5: fd %d offset %lld buffer %lx size %d Opcode %d\n",
first_ce->myaiocb64.aio_fildes,
first_ce->myaiocb64.aio_offset,
(long)(first_ce->myaiocb64.aio_buf),
first_ce->myaiocb64.aio_nbytes,
first_ce->myaiocb64.aio_lio_opcode
#endif
#else
printf("aio_return error5: fd %d offset %ld buffer %lx size %d Opcode %d\n",
first_ce->myaiocb.aio_fildes,
first_ce->myaiocb.aio_offset,
(long)(first_ce->myaiocb.aio_buf),
first_ce->myaiocb.aio_nbytes,
first_ce->myaiocb.aio_lio_opcode
#endif
);
}
if(retval > 0)
{
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
*ubuffer=(char *)first_ce->myaiocb.aio_buf;
#else
*ubuffer=(char *)first_ce->myaiocb64.aio_buf;
#endif
#else
*ubuffer=(char *)first_ce->myaiocb.aio_buf;
#endif
}else
*ubuffer=(char *)0;
first_ce->direct=1; /* do not delete the buffer */
takeoff_cache(gc,first_ce);
putoninuse(gc,first_ce);
}
return((int)retval);
}
/************************************************************************
* The caller is now finished with the data that was provided so
* the library is now free to return the memory to the pool for later
* reuse.
************************************************************************/
void
async_release(gc)
struct cache *gc;
{
takeoffinuse(gc);
}
/************************************************************************
* Put the buffer on the inuse list. When the user is finished with
* the buffer it will call back into async_release and the items on the
* inuse list will be deallocated.
************************************************************************/
void
putoninuse(gc,entry)
struct cache *gc;
struct cache_ent *entry;
{
if(gc->inuse_head)
entry->forward=gc->inuse_head;
else
entry->forward=0;
gc->inuse_head=entry;
}
/************************************************************************
* This is called when the application is finished with the data that
* was provided. The memory may now be returned to the pool.
************************************************************************/
void
takeoffinuse(gc)
struct cache *gc;
{
struct cache_ent *ce;
if(gc->inuse_head==0)
printf("Takeoffinuse error\n");
ce=gc->inuse_head;
gc->inuse_head=gc->inuse_head->forward;
if(gc->inuse_head !=0)
printf("Error in take off inuse\n");
free((void*)(ce->real_address));
free(ce);
}
/*************************************************************************
* This routine is a generic async writer assist funtion. It takes
* the same calling parameters as write() but also extends the
* interface to include:
*
* offset ..... offset in the file.
* depth ..... How much read-ahead do you want.
*
*************************************************************************/
size_t
async_write(gc,fd,buffer,size,offset,depth)
struct cache *gc;
long long fd,size;
char *buffer;
off64_t offset;
long long depth;
{
struct cache_ent *ce;
size_t ret;
ce=allocate_write_buffer(gc,fd,offset,size,(long long)LIO_WRITE,depth,0LL,(char *)0,(char *)0);
ce->direct=0; /* not direct. Lib supplies buffer and must free it */
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
mbcopy(buffer,(char *)(ce->myaiocb.aio_buf),(size_t)size);
#else
mbcopy(buffer,(char *)(ce->myaiocb64.aio_buf),(size_t)size);
#endif
#else
mbcopy(buffer,(char *)(ce->myaiocb.aio_buf),(size_t)size);
#endif
async_put_on_write_queue(gc,ce);
/*
printf("asw: fd %d offset %lld, size %d\n",ce->myaiocb64.aio_fildes,
ce->myaiocb64.aio_offset,
ce->myaiocb64.aio_nbytes);
*/
again:
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
ret=aio_write(&ce->myaiocb);
#else
ret=aio_write64(&ce->myaiocb64);
#endif
#else
ret=aio_write(&ce->myaiocb);
#endif
if(ret==-1)
{
if(errno==EAGAIN)
{
async_wait_for_write(gc);
goto again;
}
if(errno==0)
{
/* Compensate for bug in async library */
async_wait_for_write(gc);
goto again;
}
else
{
printf("Error in aio_write: ret %d errno %d count %lld\n",ret,errno,gc->w_count);
/*
printf("aio_write_no_copy: fd %d buffer %x offset %lld size %d\n",
ce->myaiocb64.aio_fildes,
ce->myaiocb64.aio_buf,
ce->myaiocb64.aio_offset,
ce->myaiocb64.aio_nbytes);
*/
exit(177);
}
}
return((ssize_t)size);
}
/*************************************************************************
* Allocate a write aiocb and write buffer of the size specified. Also
* put some extra buffer padding so that VX_DIRECT can do its job when
* needed.
*************************************************************************/
struct cache_ent *
allocate_write_buffer(gc,fd,offset,size,op,w_depth,direct,buffer,free_addr)
struct cache *gc;
long long fd,size,op;
off64_t offset;
long long w_depth;
long long direct;
char *buffer,*free_addr;
{
struct cache_ent *ce;
long temp;
if(fd==0LL)
{
printf("Setting up write buffer insane\n");
exit(178);
}
if(gc->w_count > w_depth)
async_wait_for_write(gc);
ce=(struct cache_ent *)malloc((size_t)sizeof(struct cache_ent));
if(ce == (struct cache_ent *)0)
{
printf("Malloc failed 1\n");
exit(179);
}
bzero(ce,sizeof(struct cache_ent));
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
ce->myaiocb.aio_fildes=(int)fd;
ce->myaiocb.aio_offset=(off64_t)offset;
if(!direct)
{
ce->real_address = (char *)malloc((size_t)(size+page_size));
temp=(long)ce->real_address;
temp = (temp+page_size) & ~(page_size-1);
ce->myaiocb.aio_buf=(volatile void *)temp;
}else
{
ce->myaiocb.aio_buf=(volatile void *)buffer;
ce->real_address=(char *)free_addr;
}
if(ce->myaiocb.aio_buf == 0)
#else
ce->myaiocb64.aio_fildes=(int)fd;
ce->myaiocb64.aio_offset=(off64_t)offset;
if(!direct)
{
ce->real_address = (char *)malloc((size_t)(size+page_size));
temp=(long)ce->real_address;
temp = (temp+page_size) & ~(page_size-1);
ce->myaiocb64.aio_buf=(volatile void *)temp;
}
else
{
ce->myaiocb64.aio_buf=(volatile void *)buffer;
ce->real_address=(char *)free_addr;
}
if(ce->myaiocb64.aio_buf == 0)
#endif
#else
ce->myaiocb.aio_fildes=(int)fd;
ce->myaiocb.aio_offset=(off_t)offset;
if(!direct)
{
ce->real_address = (char *)malloc((size_t)(size+page_size));
temp=(long)ce->real_address;
temp = (temp+page_size) & ~(page_size-1);
ce->myaiocb.aio_buf=(volatile void *)temp;
}
else
{
ce->myaiocb.aio_buf=(volatile void *)buffer;
ce->real_address=(char *)free_addr;
}
if(ce->myaiocb.aio_buf == 0)
#endif
{
printf("Malloc failed 2\n");
exit(180);
}
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
ce->myaiocb.aio_reqprio=0;
ce->myaiocb.aio_nbytes=(size_t)size;
ce->myaiocb.aio_sigevent.sigev_notify=SIGEV_NONE;
ce->myaiocb.aio_lio_opcode=(int)op;
#else
ce->myaiocb64.aio_reqprio=0;
ce->myaiocb64.aio_nbytes=(size_t)size;
ce->myaiocb64.aio_sigevent.sigev_notify=SIGEV_NONE;
ce->myaiocb64.aio_lio_opcode=(int)op;
#endif
#else
ce->myaiocb.aio_reqprio=0;
ce->myaiocb.aio_nbytes=(size_t)size;
ce->myaiocb.aio_sigevent.sigev_notify=SIGEV_NONE;
ce->myaiocb.aio_lio_opcode=(int)op;
#endif
ce->fd=(int)fd;
return(ce);
}
/*************************************************************************
* Put it on the outbound queue.
*************************************************************************/
void
async_put_on_write_queue(gc,ce)
struct cache *gc;
struct cache_ent *ce;
{
ce->forward=0;
ce->back=gc->w_tail;
if(gc->w_tail)
gc->w_tail->forward = ce;
gc->w_tail= ce;
if(!gc->w_head)
gc->w_head=ce;
gc->w_count++;
return;
}
/*************************************************************************
* Cleanup all outstanding writes
*************************************************************************/
void
async_write_finish(gc)
struct cache *gc;
{
while(gc->w_head)
{
/*printf("async_write_finish: Waiting for buffer %x to finish\n",gc->w_head->myaiocb64.aio_buf);*/
async_wait_for_write(gc);
}
}
/*************************************************************************
* Wait for an I/O to finish
*************************************************************************/
void
async_wait_for_write(gc)
struct cache *gc;
{
struct cache_ent *ce;
size_t ret,retval;
if(gc->w_head==0)
return;
ce=gc->w_head;
gc->w_head=ce->forward;
gc->w_count--;
ce->forward=0;
if(ce==gc->w_tail)
gc->w_tail=0;
/*printf("Wait for buffer %x offset %lld size %d to finish\n",
ce->myaiocb64.aio_buf,
ce->myaiocb64.aio_offset,
ce->myaiocb64.aio_nbytes);
printf("write count %lld \n",gc->w_count);
*/
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
while((ret=aio_error(&ce->myaiocb))== EINPROGRESS)
{
async_suspend(ce);
}
#else
while((ret=aio_error64(&ce->myaiocb64))== EINPROGRESS)
{
async_suspend(ce);
}
#endif
#else
while((ret=aio_error(&ce->myaiocb))== EINPROGRESS)
{
async_suspend(ce);
}
#endif
if(ret)
{
printf("aio_error 5: ret %d %d\n",ret,errno);
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
printf("fd %d offset %lld size %d\n",
ce->myaiocb.aio_fildes,
ce->myaiocb.aio_offset,
ce->myaiocb.aio_nbytes);
#else
printf("fd %d offset %lld size %d\n",
ce->myaiocb64.aio_fildes,
ce->myaiocb64.aio_offset,
ce->myaiocb64.aio_nbytes);
#endif
#else
printf("fd %d offset %lld size %d\n",
ce->myaiocb.aio_fildes,
ce->myaiocb.aio_offset,
ce->myaiocb.aio_nbytes);
#endif
exit(181);
}
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
retval=aio_return(&ce->myaiocb);
#else
#if defined(__CrayX1__)
retval=aio_return64((aiocb64_t *)&ce->myaiocb64);
#else
retval=aio_return64((struct aiocb64 *)&ce->myaiocb64);
#endif
#endif
#else
retval=aio_return(&ce->myaiocb);
#endif
if((int)retval < 0)
{
printf("aio_return error: %d\n",errno);
}
if(!ce->direct)
{
/* printf("Freeing buffer %x\n",ce->real_address);*/
free((void *)(ce->real_address));
free((void *)ce);
}
}
/*************************************************************************
* This routine is a generic async writer assist funtion. It takes
* the same calling parameters as write() but also extends the
* interface to include:
*
* offset ..... offset in the file.
* depth ..... How much read-ahead do you want.
* free_addr .. address of memory to free after write is completed.
*
*************************************************************************/
size_t
async_write_no_copy(gc,fd,buffer,size,offset,depth,free_addr)
struct cache *gc;
long long fd,size;
char *buffer;
off64_t offset;
long long depth;
char *free_addr;
{
struct cache_ent *ce;
size_t ret;
long long direct = 1;
ce=allocate_write_buffer(gc,fd,offset,size,(long long)LIO_WRITE,depth,direct,buffer,free_addr);
ce->direct=0; /* have library de-allocate the buffer */
async_put_on_write_queue(gc,ce);
/*
printf("awnc: fd %d offset %lld, size %d\n",ce->myaiocb64.aio_fildes,
ce->myaiocb64.aio_offset,
ce->myaiocb64.aio_nbytes);
*/
again:
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
ret=aio_write(&ce->myaiocb);
#else
ret=aio_write64(&ce->myaiocb64);
#endif
#else
ret=aio_write(&ce->myaiocb);
#endif
if(ret==-1)
{
if(errno==EAGAIN)
{
async_wait_for_write(gc);
goto again;
}
if(errno==0)
{
/* Compensate for bug in async library */
async_wait_for_write(gc);
goto again;
}
else
{
printf("Error in aio_write: ret %d errno %d\n",ret,errno);
#ifdef _LARGEFILE64_SOURCE
#ifdef __LP64__
printf("aio_write_no_copy: fd %d buffer %lx offset %lld size %d\n",
ce->myaiocb.aio_fildes,
(long)(ce->myaiocb.aio_buf),
ce->myaiocb.aio_offset,
ce->myaiocb.aio_nbytes);
#else
printf("aio_write_no_copy: fd %d buffer %lx offset %lld size %d\n",
ce->myaiocb64.aio_fildes,
(long)(ce->myaiocb64.aio_buf),
ce->myaiocb64.aio_offset,
ce->myaiocb64.aio_nbytes);
#endif
#else
printf("aio_write_no_copy: fd %d buffer %lx offset %ld size %d\n",
ce->myaiocb.aio_fildes,
(long)(ce->myaiocb.aio_buf),
ce->myaiocb.aio_offset,
ce->myaiocb.aio_nbytes);
#endif
exit(182);
}
}
else
{
return((ssize_t)size);
}
}
void mbcopy(source, dest, len)
char *source,*dest;
size_t len;
{
int i;
for(i=0;i<len;i++)
*dest++=*source++;
}