开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > 数据库系统 > MySQL数据库 > 查看源码
AsyncFile.cpp
资源名称：mysql-4.1.16-win-src.zip　[点击查看]
上传用户：romrleung
上传日期：2022-05-23
资源大小：18897k
文件大小：24k
源码类别：

MySQL数据库

开发平台：

Visual C++

AsyncFile.cpp：源码内容
  1. /* Copyright (C) 2003 MySQL AB
  3.    This program is free software; you can redistribute it and/or modify
  4.    it under the terms of the GNU General Public License as published by
  5.    the Free Software Foundation; either version 2 of the License, or
  6.    (at your option) any later version.
  8.    This program is distributed in the hope that it will be useful,
  9.    but WITHOUT ANY WARRANTY; without even the implied warranty of
  10.    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11.    GNU General Public License for more details.
  13.    You should have received a copy of the GNU General Public License
  14.    along with this program; if not, write to the Free Software
  15.    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
  17. #include <ndb_global.h>
  18. #include <my_sys.h>
  19. #include <my_pthread.h>
  21. #include <Error.hpp>
  22. #include "AsyncFile.hpp"
  24. #include <ErrorHandlingMacros.hpp>
  25. #include <kernel_types.h>
  26. #include <NdbMem.h>
  27. #include <NdbThread.h>
  28. #include <signaldata/FsOpenReq.hpp>
  30. // use this to test broken pread code
  31. //#define HAVE_BROKEN_PREAD 
  33. #ifdef HAVE_BROKEN_PREAD
  34. #undef HAVE_PWRITE
  35. #undef HAVE_PREAD
  36. #endif
  38. #if defined NDB_WIN32 || defined NDB_OSE || defined NDB_SOFTOSE
  39. #else
  40. // For readv and writev
  41. #include <sys/uio.h> 
  42. #endif
  44. #ifndef NDB_WIN32
  45. #include <dirent.h>
  46. #endif
  48. // Use this define if you want printouts from AsyncFile class
  49. //#define DEBUG_ASYNCFILE
  51. #ifdef DEBUG_ASYNCFILE
  52. #include <NdbOut.hpp>
  53. #define DEBUG(x) x
  54. #define PRINT_ERRORANDFLAGS(f) printErrorAndFlags(f)
  55. void printErrorAndFlags(Uint32 used_flags);
  56. #else
  57. #define DEBUG(x) 
  58. #define PRINT_ERRORANDFLAGS(f)
  59. #endif
  61. // Define the size of the write buffer (for each thread)
  62. #if defined NDB_SOFTOSE || defined NDB_OSE
  63. #define WRITEBUFFERSIZE 65536
  64. #else 
  65. #define WRITEBUFFERSIZE 262144
  66. #endif
  68. const char *actionName[] = {
  69.   "open",
  70.   "close",
  71.   "closeRemove",
  72.   "read",
  73.   "readv",
  74.   "write",
  75.   "writev",
  76.   "writeSync",
  77.   "writevSync",
  78.   "sync",
  79.   "end" };
  81. static int numAsyncFiles = 0;
  83. extern "C" void * runAsyncFile(void* arg)
  84. {
  85.   ((AsyncFile*)arg)->run();
  86.   return (NULL);
  87. }
  89. AsyncFile::AsyncFile() :
  90.   theFileName(),
  91. #ifdef NDB_WIN32
  92.   hFile(INVALID_HANDLE_VALUE),
  93. #else
  94.   theFd(-1),
  95. #endif
  96.   theReportTo(0),
  97.   theMemoryChannelPtr(NULL)
  98. {
  99. }
  101. void
  102. AsyncFile::doStart(Uint32 nodeId,
  103.    const char * filesystemPath,
  104.    const char * backup_path) {
  105.   theFileName.init(nodeId, filesystemPath, backup_path);
  107.   // Stacksize for filesystem threads
  108.   // An 8k stack should be enough
  109.   const NDB_THREAD_STACKSIZE stackSize = 8192;
  111.   char buf[16];
  112.   numAsyncFiles++;
  113.   BaseString::snprintf(buf, sizeof(buf), "AsyncFile%d", numAsyncFiles);
  115.   theStartMutexPtr = NdbMutex_Create();
  116.   theStartConditionPtr = NdbCondition_Create();
  117.   NdbMutex_Lock(theStartMutexPtr);
  118.   theStartFlag = false;
  119.   theThreadPtr = NdbThread_Create(runAsyncFile,
  120.                                   (void**)this,
  121.                                   stackSize,
  122.                                   (char*)&buf,
  123.                                   NDB_THREAD_PRIO_MEAN);
  125.   NdbCondition_Wait(theStartConditionPtr,
  126.                     theStartMutexPtr);    
  127.   NdbMutex_Unlock(theStartMutexPtr);
  128.   NdbMutex_Destroy(theStartMutexPtr);
  129.   NdbCondition_Destroy(theStartConditionPtr);
  130. }
  132. AsyncFile::~AsyncFile() 
  133. {
  134.   void *status;
  135.   Request request;
  136.   request.action = Request::end;
  137.   theMemoryChannelPtr->writeChannel( &request );
  138.   NdbThread_WaitFor(theThreadPtr, &status);
  139.   NdbThread_Destroy(&theThreadPtr);
  140.   delete theMemoryChannelPtr;
  141. }
  143. void
  144. AsyncFile::reportTo( MemoryChannel<Request> *reportTo )
  145. {
  146.   theReportTo = reportTo;
  147. }
  149. void AsyncFile::execute(Request* request) 
  150. {
  151.   theMemoryChannelPtr->writeChannel( request );
  152. }
  154. void
  155. AsyncFile::run()
  156. {
  157.   Request *request;
  158.   // Create theMemoryChannel in the thread that will wait for it
  159.   NdbMutex_Lock(theStartMutexPtr);
  160.   theMemoryChannelPtr = new MemoryChannel<Request>();
  161.   theStartFlag = true;
  162.   // Create write buffer for bigger writes
  163.   theWriteBufferSize = WRITEBUFFERSIZE;
  164.   theWriteBuffer = (char *) NdbMem_Allocate(theWriteBufferSize); 
  165.   NdbMutex_Unlock(theStartMutexPtr);
  166.   NdbCondition_Signal(theStartConditionPtr);
  167.   
  168.   if (!theWriteBuffer) {
  169.     DEBUG(ndbout_c("AsyncFile::writeReq, Failed allocating write buffer"));
  170.     return;
  171.   }//if
  173.   while (1) {
  174.     request = theMemoryChannelPtr->readChannel();
  175.     if (!request) {
  176.       DEBUG(ndbout_c("Nothing read from Memory Channel in AsyncFile"));
  177.       endReq();
  178.       return;
  179.     }//if
  180.     switch (request->action) {
  181.     case Request:: open:
  182.       openReq(request);
  183.       break;
  184.     case Request:: close:
  185.       closeReq(request);
  186.       break;
  187.     case Request:: closeRemove:
  188.       closeReq(request);
  189.       removeReq(request);
  190.       break;
  191.     case Request:: read:
  192.       readReq(request);
  193.       break;
  194.     case Request:: readv:
  195.       readvReq(request);
  196.       break;
  197.     case Request:: write:
  198.       writeReq(request);
  199.       break;
  200.     case Request:: writev:
  201.       writevReq(request);
  202.       break;
  203.     case Request:: writeSync:
  204.       writeReq(request);
  205.       syncReq(request);
  206.       break;
  207.     case Request:: writevSync:
  208.       writevReq(request);
  209.       syncReq(request);
  210.       break;
  211.     case Request:: sync:
  212.       syncReq(request);
  213.       break;
  214.     case Request:: append:
  215.       appendReq(request);
  216.       break;
  217.     case Request::rmrf:
  218.       rmrfReq(request, (char*)theFileName.c_str(), request->par.rmrf.own_directory);
  219.       break;
  220.     case Request:: end:
  221.       if (theFd > 0)
  222.         closeReq(request);
  223.       endReq();
  224.       return;
  225.     default:
  226.       abort();
  227.       break;
  228.     }//switch
  229.     
  230.     // No need to signal as ndbfs only uses tryRead
  231.     theReportTo->writeChannelNoSignal(request);
  232.   }//while
  233. }//AsyncFile::run()
  235. extern bool Global_useO_SYNC;
  236. extern bool Global_useO_DIRECT;
  237. extern bool Global_unlinkO_CREAT;
  238. extern Uint32 Global_syncFreq;
  240. void AsyncFile::openReq(Request* request)
  241. {  
  242.   m_openedWithSync = false;
  243.   m_syncFrequency = 0;
  244.   m_syncCount= 0;
  246.   // for open.flags, see signal FSOPENREQ
  247. #ifdef NDB_WIN32
  248.   DWORD dwCreationDisposition;
  249.   DWORD dwDesiredAccess = 0;
  250.   DWORD dwShareMode = FILE_SHARE_READ | FILE_SHARE_WRITE;
  251.   DWORD dwFlagsAndAttributes = FILE_ATTRIBUTE_NORMAL | FILE_FLAG_RANDOM_ACCESS | FILE_FLAG_NO_BUFFERING;
  252.   const Uint32 flags = request->par.open.flags;
  253.     
  254.     // Convert file open flags from Solaris to Windows
  255.   if ((flags & FsOpenReq::OM_CREATE) && (flags & FsOpenReq::OM_TRUNCATE)){
  256.     dwCreationDisposition = CREATE_ALWAYS;
  257.   } else if (flags & FsOpenReq::OM_TRUNCATE){
  258.     dwCreationDisposition = TRUNCATE_EXISTING;
  259.   } else if (flags & FsOpenReq::OM_CREATE){
  260.     dwCreationDisposition = CREATE_NEW;
  261.   } else {
  262.     dwCreationDisposition = OPEN_EXISTING;
  263.   }
  264.   
  265.   switch(flags & 3){
  266.   case FsOpenReq::OM_READONLY:
  267.     dwDesiredAccess = GENERIC_READ;
  268.     break;
  269.   case FsOpenReq::OM_WRITEONLY:
  270.     dwDesiredAccess = GENERIC_WRITE;
  271.     break;
  272.   case FsOpenReq::OM_READWRITE:
  273.     dwDesiredAccess = GENERIC_READ | GENERIC_WRITE;
  274.     break;
  275.   default:
  276.     request->error = 1000;
  277.     break;
  278.     return;
  279.   }
  281.   hFile = CreateFile(theFileName.c_str(), dwDesiredAccess, dwShareMode, 
  282.                      0, dwCreationDisposition, dwFlagsAndAttributes, 0);
  283.     
  284.   if(INVALID_HANDLE_VALUE == hFile) {
  285.     request->error = GetLastError();
  286.     if(((ERROR_PATH_NOT_FOUND == request->error) || (ERROR_INVALID_NAME == request->error))
  287.        && (flags & FsOpenReq::OM_CREATE)) {
  288.       createDirectories();
  289.       hFile = CreateFile(theFileName.c_str(), dwDesiredAccess, dwShareMode, 
  290.                          0, dwCreationDisposition, dwFlagsAndAttributes, 0);
  291.             
  292.       if(INVALID_HANDLE_VALUE == hFile)
  293.         request->error = GetLastError();
  294.       else
  295.         request->error = 0;
  296.             
  297.       return;
  298.     }
  299.   } 
  300.   else {
  301.     request->error = 0;
  302.     return;
  303.   }
  304. #else
  305.   const Uint32 flags = request->par.open.flags;
  306.   Uint32 new_flags = 0;
  308.   // Convert file open flags from Solaris to Liux
  309.   if(flags & FsOpenReq::OM_CREATE){
  310.     new_flags |= O_CREAT;
  311.   }
  313.   if(flags & FsOpenReq::OM_TRUNCATE){
  314. #if 0
  315.     if(Global_unlinkO_CREAT){
  316.       unlink(theFileName.c_str());
  317.     } else 
  318. #endif
  319.       new_flags |= O_TRUNC;
  320.   }  
  322.   if(flags & FsOpenReq::OM_APPEND){
  323.     new_flags |= O_APPEND;
  324.   }
  326.   if(flags & FsOpenReq::OM_SYNC){
  327. #if 0
  328.     if(Global_useO_SYNC){
  329.       new_flags |= O_SYNC;
  330.       m_openedWithSync = true;
  331.       m_syncFrequency = 0;
  332.     } else {
  333. #endif
  334.       m_openedWithSync = false;
  335.       m_syncFrequency = Global_syncFreq;
  336. #if 0
  337.     }
  338. #endif
  339.   } else {
  340.     m_openedWithSync = false;
  341.     m_syncFrequency = 0;
  342.   }
  344. #if 0
  345.   //#if NDB_LINUX
  346.   if(Global_useO_DIRECT){
  347.     new_flags |= O_DIRECT;
  348.   }
  349. #endif
  351.   switch(flags & 0x3){
  352.   case FsOpenReq::OM_READONLY:
  353.     new_flags |= O_RDONLY;
  354.     break;
  355.   case FsOpenReq::OM_WRITEONLY:
  356.     new_flags |= O_WRONLY;
  357.     break;
  358.   case FsOpenReq::OM_READWRITE:
  359.     new_flags |= O_RDWR;
  360.     break;
  361.   default:
  362.     request->error = 1000;
  363.     break;
  364.     return;
  365.   }
  366.   const int mode = S_IRUSR | S_IWUSR | S_IRGRP;
  367.   
  368.   if (-1 == (theFd = ::open(theFileName.c_str(), new_flags, mode))) {
  369.     PRINT_ERRORANDFLAGS(new_flags);
  370.     if( (errno == ENOENT ) && (new_flags & O_CREAT ) ) {
  371.       createDirectories();
  372.       if (-1 == (theFd = ::open(theFileName.c_str(), new_flags, mode))) {
  373.         PRINT_ERRORANDFLAGS(new_flags);
  374.         request->error = errno;
  375.       }
  376.     } else {
  377.       request->error = errno;
  378.     }
  379.   }
  380. #endif
  381. }
  383. int
  384. AsyncFile::readBuffer(char * buf, size_t size, off_t offset){
  385.   int return_value;
  386.   
  387. #ifdef NDB_WIN32
  388.   DWORD dwSFP = SetFilePointer(hFile, offset, 0, FILE_BEGIN);
  389.   if(dwSFP != offset) {
  390.     return GetLastError();
  391.   }
  392. #elif ! defined(HAVE_PREAD)
  393.   off_t seek_val;
  394.   while((seek_val= lseek(theFd, offset, SEEK_SET)) == (off_t)-1 
  395. && errno == EINTR);
  396.   if(seek_val == (off_t)-1)
  397.   {
  398.     return errno;
  399.   }
  400. #endif
  401.     
  402.   while (size > 0) {
  403.     size_t bytes_read = 0;
  404.     
  405. #ifdef NDB_WIN32
  406.     DWORD dwBytesRead;
  407.     BOOL bRead = ReadFile(hFile, 
  408.                           buf,
  409.                           size,
  410.                           &dwBytesRead,
  411.                           0);
  412.     if(!bRead){
  413.       return GetLastError();
  414.     } 
  415.     bytes_read = dwBytesRead;
  416. #elif  ! defined(HAVE_PREAD)
  417.     return_value = ::read(theFd, buf, size);
  418. #else // UNIX
  419.     return_value = ::pread(theFd, buf, size, offset);
  420. #endif
  421. #ifndef NDB_WIN32
  422.     if (return_value == -1 && errno == EINTR) {
  423.       DEBUG(ndbout_c("EINTR in read"));
  424.       continue;
  425.     } else if (return_value == -1){
  426.       return errno;
  427.     } else {
  428.       bytes_read = return_value;
  429.     }
  430. #endif
  432.     if(bytes_read == 0){
  433.       DEBUG(ndbout_c("Read underflow %d %dn %xn%d %d", 
  434.             size, offset, buf, bytes_read, return_value));
  435.       return ERR_ReadUnderflow;
  436.     }
  437.     
  438.     if(bytes_read != size){
  439.       DEBUG(ndbout_c("Warning partial read %d != %d", 
  440.             bytes_read, size));
  441.     }
  443.     buf += bytes_read;
  444.     size -= bytes_read;
  445.     offset += bytes_read;
  446.   }
  447.   return 0;
  448. }
  450. void
  451. AsyncFile::readReq( Request * request)
  452. {
  453.   for(int i = 0; i < request->par.readWrite.numberOfPages ; i++) {
  454.     off_t offset = request->par.readWrite.pages[i].offset;
  455.     size_t size  = request->par.readWrite.pages[i].size;
  456.     char * buf   = request->par.readWrite.pages[i].buf;
  457.     
  458.     int err = readBuffer(buf, size, offset);
  459.     if(err != 0){
  460.       request->error = err;
  461.       return;
  462.     }
  463.   }
  464. }
  466. void
  467. AsyncFile::readvReq( Request * request)
  468. {
  469. #if ! defined(HAVE_PREAD)
  470.   readReq(request);
  471.   return;
  472. #elif defined NDB_WIN32
  473.   // ReadFileScatter?
  474.   readReq(request);
  475.   return;
  476. #else
  477.   int return_value;
  478.   int length = 0;
  479.   struct iovec iov[20]; // the parameter in the signal restricts this to 20 deep
  480.   for(int i=0; i < request->par.readWrite.numberOfPages ; i++) {
  481.     iov[i].iov_base= request->par.readWrite.pages[i].buf;
  482.     iov[i].iov_len= request->par.readWrite.pages[i].size;
  483.     length = length + iov[i].iov_len;
  484.   }
  485.   lseek( theFd, request->par.readWrite.pages[0].offset, SEEK_SET );
  486.   return_value = ::readv(theFd, iov, request->par.readWrite.numberOfPages);
  487.   if (return_value == -1) {
  488.     request->error = errno;
  489.     return;
  490.   } else if (return_value != length) {
  491.     request->error = 1011;
  492.     return;
  493.   }
  494. #endif
  495. }
  497. int 
  498. AsyncFile::extendfile(Request* request) {
  499. #if ! defined(HAVE_PWRITE)
  500.   // Find max size of this file in this request
  501.   int maxOffset = 0;
  502.   int maxSize = 0;
  503.   for(int i=0; i < request->par.readWrite.numberOfPages ; i++) {
  504.     if (request->par.readWrite.pages[i].offset > maxOffset) {
  505.       maxOffset = request->par.readWrite.pages[i].offset;
  506.       maxSize = request->par.readWrite.pages[i].size;
  507.     }
  508.   }
  509.   DEBUG(ndbout_c("extendfile: maxOffset=%d, size=%d", maxOffset, maxSize));
  511.   // Allocate a buffer and fill it with zeros
  512.   void* pbuf = NdbMem_Allocate(maxSize);
  513.   memset(pbuf, 0, maxSize);
  514.   for (int p = 0; p <= maxOffset; p = p + maxSize) {
  515.     int return_value;
  516.     return_value = lseek(theFd, 
  517.                          p,
  518.                          SEEK_SET);
  519.     if((return_value == -1 ) || (return_value != p)) {
  520.       return -1;
  521.     }
  522.     return_value = ::write(theFd, 
  523.                            pbuf,
  524.                            maxSize);
  525.     if ((return_value == -1) || (return_value != maxSize)) {
  526.       return -1;
  527.     }
  528.   }
  529.   free(pbuf);
  530.   
  531.   DEBUG(ndbout_c("extendfile: "%s" OK!", theFileName.c_str()));
  532.   return 0;
  533. #else
  534.   request = request;
  535.   abort();
  536.   return -1;
  537. #endif
  538. }
  540. void
  541. AsyncFile::writeReq( Request * request)
  542. {
  543.   int page_num = 0;
  544.   bool write_not_complete = true;
  546.   while(write_not_complete) {
  547.     int totsize = 0;
  548.     off_t offset = request->par.readWrite.pages[page_num].offset;
  549.     char* bufptr = theWriteBuffer;
  550.       
  551.     write_not_complete = false;
  552.     if (request->par.readWrite.numberOfPages > 1) {
  553.       off_t page_offset = offset;
  555.       // Multiple page write, copy to buffer for one write
  556.       for(int i=page_num; i < request->par.readWrite.numberOfPages; i++) {
  557.         memcpy(bufptr, 
  558.                request->par.readWrite.pages[i].buf, 
  559.                request->par.readWrite.pages[i].size);
  560.         bufptr += request->par.readWrite.pages[i].size;
  561.         totsize += request->par.readWrite.pages[i].size;
  562.         if (((i + 1) < request->par.readWrite.numberOfPages)) {
  563.           // There are more pages to write
  564.           // Check that offsets are consequtive
  565.   off_t tmp = page_offset + request->par.readWrite.pages[i].size;
  566.           if (tmp != request->par.readWrite.pages[i+1].offset) {
  567.             // Next page is not aligned with previous, not allowed
  568.             DEBUG(ndbout_c("Page offsets are not aligned"));
  569.             request->error = EINVAL;
  570.             return;
  571.           }
  572.           if ((unsigned)(totsize + request->par.readWrite.pages[i+1].size) > (unsigned)theWriteBufferSize) {
  573.             // We are not finished and the buffer is full
  574.             write_not_complete = true;
  575.             // Start again with next page
  576.             page_num = i + 1;
  577.             break;
  578.           }
  579.         }
  580.         page_offset += request->par.readWrite.pages[i].size;
  581.       }
  582.       bufptr = theWriteBuffer;
  583.     } else { 
  584.       // One page write, write page directly
  585.       bufptr = request->par.readWrite.pages[0].buf;
  586.       totsize = request->par.readWrite.pages[0].size;
  587.     }
  588.     int err = writeBuffer(bufptr, totsize, offset);
  589.     if(err != 0){
  590.       request->error = err;
  591.       return;
  592.     }
  593.   } // while(write_not_complete)
  594. }
  596. int
  597. AsyncFile::writeBuffer(const char * buf, size_t size, off_t offset, 
  598.        size_t chunk_size)
  599. {
  600.   size_t bytes_to_write = chunk_size;
  601.   int return_value;
  603. #ifdef NDB_WIN32
  604.   DWORD dwSFP = SetFilePointer(hFile, offset, 0, FILE_BEGIN);
  605.   if(dwSFP != offset) {
  606.     return GetLastError();
  607.   }
  608. #elif ! defined(HAVE_PWRITE)
  609.   off_t seek_val;
  610.   while((seek_val= lseek(theFd, offset, SEEK_SET)) == (off_t)-1 
  611. && errno == EINTR);
  612.   if(seek_val == (off_t)-1)
  613.   {
  614.     return errno;
  615.   }
  616. #endif
  617.     
  618.   while (size > 0) {
  619.     if (size < bytes_to_write){
  620.       // We are at the last chunk
  621.       bytes_to_write = size;
  622.     }
  623.     size_t bytes_written = 0;
  624.     
  625. #ifdef NDB_WIN32
  626.     DWORD dwWritten;
  627.     BOOL bWrite = WriteFile(hFile, buf, bytes_to_write, &dwWritten, 0);
  628.     if(!bWrite) {
  629.       return GetLastError();
  630.     }
  631.     bytes_written = dwWritten;
  632.     if (bytes_written != bytes_to_write) {
  633.       DEBUG(ndbout_c("Warning partial write %d != %d", bytes_written, bytes_to_write));
  634.     }
  635.     
  636. #elif ! defined(HAVE_PWRITE)
  637.     return_value = ::write(theFd, buf, bytes_to_write);
  638. #else // UNIX
  639.     return_value = ::pwrite(theFd, buf, bytes_to_write, offset);
  640. #endif
  641. #ifndef NDB_WIN32
  642.     if (return_value == -1 && errno == EINTR) {
  643.       bytes_written = 0;
  644.       DEBUG(ndbout_c("EINTR in write"));
  645.     } else if (return_value == -1){
  646.       return errno;
  647.     } else {
  648.       bytes_written = return_value;
  650.       if(bytes_written == 0){
  651. abort();
  652.       }
  653.       
  654.       if(bytes_written != bytes_to_write){
  655. DEBUG(ndbout_c("Warning partial write %d != %d", 
  656.  bytes_written, bytes_to_write));
  657.       }
  658.     }
  659. #endif
  660.     
  661.     m_syncCount+= bytes_written;
  662.     buf += bytes_written;
  663.     size -= bytes_written;
  664.     offset += bytes_written;
  665.   }
  666.   return 0;
  667. }
  669. void
  670. AsyncFile::writevReq( Request * request)
  671. {
  672.   // WriteFileGather on WIN32?
  673.   writeReq(request);
  674. }
  677. void
  678. AsyncFile::closeReq(Request * request)
  679. {
  680.   syncReq(request);
  681. #ifdef NDB_WIN32
  682.   if(!CloseHandle(hFile)) {
  683.     request->error = GetLastError();
  684.   }
  685.   hFile = INVALID_HANDLE_VALUE;
  686. #else
  687.   if (-1 == ::close(theFd)) {
  688. #ifndef DBUG_OFF
  689.     if (theFd == -1)
  690.       abort();
  691. #endif
  692.     request->error = errno;
  693.   }
  694.   theFd = -1;
  695. #endif
  696. }
  698. bool AsyncFile::isOpen(){
  699. #ifdef NDB_WIN32
  700.   return (hFile != INVALID_HANDLE_VALUE);
  701. #else
  702.   return (theFd != -1);
  703. #endif
  704. }
  707. void
  708. AsyncFile::syncReq(Request * request)
  709. {
  710.   if(m_openedWithSync ||
  711.      m_syncCount == 0){
  712.     return;
  713.   }
  714. #ifdef NDB_WIN32
  715.   if(!FlushFileBuffers(hFile)) {
  716.     request->error = GetLastError();
  717.     return;
  718.   }
  719. #else
  720.   if (-1 == ::fsync(theFd)){
  721.     request->error = errno;
  722.     return;
  723.   }
  724. #endif
  725.   m_syncCount = 0;
  726. }
  728. void
  729. AsyncFile::appendReq(Request * request){
  730.   
  731.   const char * buf = request->par.append.buf;
  732.   Uint32 size = request->par.append.size;
  734.   m_syncCount += size;
  736. #ifdef NDB_WIN32
  737.   DWORD dwWritten = 0;  
  738.   while(size > 0){
  739.     if(!WriteFile(hFile, buf, size, &dwWritten, 0)){
  740.       request->error = GetLastError();
  741.       return ;
  742.     }
  743.     
  744.     buf += dwWritten;
  745.     size -= dwWritten;
  746.   }
  747. #else
  748.   while(size > 0){
  749.     const int n = write(theFd, buf, size);
  750.     if(n == -1 && errno == EINTR){
  751.       continue;
  752.     }
  753.     if(n == -1){
  754.       request->error = errno;
  755.       return;
  756.     }
  757.     if(n == 0){
  758.       abort();
  759.     }
  760.     size -= n;
  761.     buf += n;
  762.   }
  763. #endif
  765.   if(m_syncFrequency != 0 && m_syncCount > m_syncFrequency){
  766.     syncReq(request);
  767.   }
  768. }
  770. void
  771. AsyncFile::removeReq(Request * request)
  772. {
  773. #ifdef NDB_WIN32
  774.   if(!DeleteFile(theFileName.c_str())) {
  775.     request->error = GetLastError();
  776.   }
  777. #else
  778.   if (-1 == ::remove(theFileName.c_str())) {
  779.     request->error = errno;
  781.   }
  782. #endif
  783. }
  785. void
  786. AsyncFile::rmrfReq(Request * request, char * path, bool removePath){
  787.   Uint32 path_len = strlen(path);
  788.   Uint32 path_max_copy = PATH_MAX - path_len;
  789.   char* path_add = &path[path_len];
  790. #ifndef NDB_WIN32
  791.   if(!request->par.rmrf.directory){
  792.     // Remove file
  793.     if(unlink((const char *)path) != 0 && errno != ENOENT)
  794.       request->error = errno;
  795.     return;
  796.   }
  797.   // Remove directory 
  798.   DIR* dirp = opendir((const char *)path);
  799.   if(dirp == 0){
  800.     if(errno != ENOENT) 
  801.       request->error = errno;
  802.     return;
  803.   }
  804.   struct dirent * dp;
  805.   while ((dp = readdir(dirp)) != NULL){
  806.     if ((strcmp(".", dp->d_name) != 0) && (strcmp("..", dp->d_name) != 0)) {
  807.       BaseString::snprintf(path_add, (size_t)path_max_copy, "%s%s",
  808.        DIR_SEPARATOR, dp->d_name);
  809.       if(remove((const char*)path) == 0){
  810.         path[path_len] = 0;
  811. continue;
  812.       }
  813.       
  814.       rmrfReq(request, path, true);
  815.       path[path_len] = 0;
  816.       if(request->error != 0){
  817. closedir(dirp);
  818. return;
  819.       }
  820.     }
  821.   }
  822.   closedir(dirp);
  823.   if(removePath && rmdir((const char *)path) != 0){
  824.     request->error = errno;
  825.   }
  826.   return;
  827. #else
  829.   if(!request->par.rmrf.directory){
  830.     // Remove file
  831.     if(!DeleteFile(path)){
  832.       DWORD dwError = GetLastError();
  833.       if(dwError!=ERROR_FILE_NOT_FOUND)
  834. request->error = dwError;
  835.     }
  836.     return;
  837.   }
  838.   
  839.   strcat(path, "\*");
  840.   WIN32_FIND_DATA ffd;
  841.   HANDLE hFindFile = FindFirstFile(path, &ffd);
  842.   path[path_len] = 0;
  843.   if(INVALID_HANDLE_VALUE==hFindFile){
  844.     DWORD dwError = GetLastError();
  845.     if(dwError!=ERROR_PATH_NOT_FOUND)
  846.       request->error = dwError;
  847.     return;
  848.   }
  849.   
  850.   do {
  851.     if(0!=strcmp(".", ffd.cFileName) && 0!=strcmp("..", ffd.cFileName)){
  852.       strcat(path, "\");
  853.       strcat(path, ffd.cFileName);
  854.       if(DeleteFile(path)) {
  855.         path[path_len] = 0;
  856. continue;
  857.       }//if
  858.       
  859.       rmrfReq(request, path, true);
  860.       path[path_len] = 0;
  861.       if(request->error != 0){
  862. FindClose(hFindFile);
  863. return;
  864.       }
  865.     }
  866.   } while(FindNextFile(hFindFile, &ffd));
  867.   
  868.   FindClose(hFindFile);
  869.   
  870.   if(removePath && !RemoveDirectory(path))
  871.     request->error = GetLastError();
  872.   
  873. #endif
  874. }
  876. void AsyncFile::endReq()
  877. {
  878.   // Thread is ended with return
  879.   if (theWriteBuffer) NdbMem_Free(theWriteBuffer);
  880. }
  883. void AsyncFile::createDirectories()
  884. {
  885.   for (int i = 0; i < theFileName.levels(); i++) {
  886. #ifdef NDB_WIN32
  887.     CreateDirectory(theFileName.directory(i), 0);
  888. #else
  889.     //printf("AsyncFile::createDirectories : "%s"n", theFileName.directory(i)); 
  890.     mkdir(theFileName.directory(i), S_IRUSR | S_IWUSR | S_IXUSR | S_IXGRP | S_IRGRP);
  891. #endif
  892.   }
  893. }
  895. #ifdef DEBUG_ASYNCFILE
  896. void printErrorAndFlags(Uint32 used_flags) {
  897.   char buf[255];
  898.   sprintf(buf, "PEAF: errno=%d "", errno);
  900.   switch(errno) {
  901.   case     EACCES:
  902.     strcat(buf, "EACCES");
  903.     break;
  904.   case     EDQUOT: 
  905.     strcat(buf, "EDQUOT");
  906.     break;
  907.   case     EEXIST    :
  908.     strcat(buf, "EEXIST");
  909.     break;
  910.   case     EINTR     :
  911.     strcat(buf, "EINTR");
  912.     break;
  913.   case     EFAULT    :
  914.     strcat(buf, "EFAULT");
  915.     break;
  916.   case     EIO       :
  917.     strcat(buf, "EIO");
  918.     break;
  919.   case     EISDIR    :
  920.     strcat(buf, "EISDIR");
  921.     break;
  922.   case     ELOOP       :
  923.     strcat(buf, "ELOOP");
  924.     break;
  925.   case     EMFILE   :
  926.     strcat(buf, "EMFILE");
  927.     break;
  928.   case     ENFILE    :
  929.     strcat(buf, "ENFILE");
  930.     break;
  931.   case     ENOENT    :
  932.     strcat(buf, "ENOENT ");
  933.     break;
  934.   case     ENOSPC    :
  935.     strcat(buf, "ENOSPC");
  936.     break;
  937.   case     ENOTDIR  :
  938.     strcat(buf, "ENOTDIR");
  939.     break;
  940.   case     ENXIO     : 
  941.     strcat(buf, "ENXIO");
  942.     break;  
  943.   case     EOPNOTSUPP:
  944.     strcat(buf, "EOPNOTSUPP");
  945.     break;
  946. #if !defined NDB_OSE && !defined NDB_SOFTOSE
  947.   case     EMULTIHOP :
  948.     strcat(buf, "EMULTIHOP");
  949.     break;
  950.   case     ENOLINK    :
  951.     strcat(buf, "ENOLINK");
  952.     break;
  953.   case     ENOSR      :
  954.     strcat(buf, "ENOSR");
  955.     break;      
  956.   case     EOVERFLOW :
  957.     strcat(buf,  "EOVERFLOW");
  958.     break;
  959. #endif
  960.   case     EROFS    :
  961.     strcat(buf,  "EROFS");
  962.     break;
  963.   case     EAGAIN :   
  964.     strcat(buf,  "EAGAIN");
  965.     break;
  966.   case     EINVAL    :
  967.     strcat(buf,  "EINVAL");
  968.     break;
  969.   case     ENOMEM    :
  970.     strcat(buf, "ENOMEM");
  971.     break;
  972.   case     ETXTBSY   :
  973.     strcat(buf, "ETXTBSY");
  974.     break;
  975.   case ENAMETOOLONG:
  976.     strcat(buf, "ENAMETOOLONG");
  977.     break;
  978.   case EBADF:
  979.     strcat(buf, "EBADF");
  980.     break;
  981.   case ESPIPE:
  982.     strcat(buf, "ESPIPE");
  983.     break;
  984.   case ESTALE:
  985.     strcat(buf, "ESTALE");
  986.     break;
  987.   default:  
  988.     strcat(buf, "EOTHER");
  989.     break;
  990.   }
  991.   strcat(buf, "" ");
  992. #if defined NDB_OSE
  993.   strcat(buf, strerror(errno) << " ");
  994. #endif
  995.   strcat(buf, " flags: ");
  996.   switch(used_flags & 3){
  997.   case O_RDONLY:
  998.     strcat(buf, "O_RDONLY, ");
  999.     break;
  1000.   case O_WRONLY:
  1001.     strcat(buf, "O_WRONLY, ");
  1002.     break;
  1003.   case O_RDWR:
  1004.     strcat(buf, "O_RDWR, ");
  1005.     break;
  1006.   default:
  1007.     strcat(buf, "Unknown!!, ");
  1008.   }
  1010.   if((used_flags & O_APPEND)==O_APPEND)
  1011.     strcat(buf, "O_APPEND, ");
  1012.   if((used_flags & O_CREAT)==O_CREAT)
  1013.     strcat(buf, "O_CREAT, ");
  1014.   if((used_flags & O_EXCL)==O_EXCL)
  1015.     strcat(buf, "O_EXCL, ");
  1016.   if((used_flags & O_NOCTTY) == O_NOCTTY)
  1017.     strcat(buf, "O_NOCTTY, ");
  1018.   if((used_flags & O_NONBLOCK)==O_NONBLOCK)
  1019.     strcat(buf, "O_NONBLOCK, ");
  1020.   if((used_flags & O_TRUNC)==O_TRUNC)
  1021.     strcat(buf, "O_TRUNC, ");
  1022. #if !defined NDB_OSE && !defined NDB_SOFTOSE
  1023.   if((used_flags & O_DSYNC)==O_DSYNC)
  1024.     strcat(buf, "O_DSYNC, ");
  1025.   if((used_flags & O_NDELAY)==O_NDELAY)
  1026.     strcat(buf, "O_NDELAY, ");
  1027.   if((used_flags & O_RSYNC)==O_RSYNC)
  1028.     strcat(buf, "O_RSYNC, ");
  1029.   if((used_flags & O_SYNC)==O_SYNC)
  1030.     strcat(buf, "O_SYNC, ");
  1031.   DEBUG(ndbout_c(buf));
  1032. #endif
  1034. }
  1035. #endif