← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to mysys/mf_iocache.cc

  • Committer: Monty Taylor
  • Date: 2008-11-16 06:29:53 UTC
  • mto: (584.1.9 devel)
  • mto: This revision was merged to the branch mainline in revision 589.
  • Revision ID: monty@inaugust.com-20081116062953-ivdltjmfe009b5fr
Moved stuff into item/

Show diffs side-by-side

added added

removed removed

Lines of Context:
mysys/mf_iocache.cc
47
47
  write buffer to the read buffer before we start to reuse it.
48
48
*/
49
49
 
50
 
#include "config.h"
51
 
 
52
 
#include "drizzled/internal/my_sys.h"
53
 
#include "drizzled/internal/m_string.h"
 
50
#include "mysys_priv.h"
 
51
#include <mystrings/m_string.h>
54
52
#ifdef HAVE_AIOWAIT
55
 
#include "drizzled/error.h"
56
 
#include "drizzled/internal/aio_result.h"
 
53
#include "mysys_err.h"
 
54
#include <mysys/aio_result.h>
57
55
static void my_aiowait(my_aio_result *result);
58
56
#endif
59
 
#include "drizzled/internal/iocache.h"
 
57
#include <mysys/iocache.h>
60
58
#include <errno.h>
61
59
#include <drizzled/util/test.h>
62
 
#include <stdlib.h>
63
 
#include <algorithm>
64
 
 
65
 
using namespace std;
66
 
 
67
 
namespace drizzled
68
 
{
69
 
namespace internal
70
 
{
71
 
 
72
 
static int _my_b_read(register IO_CACHE *info, unsigned char *Buffer, size_t Count);
73
 
static int _my_b_write(register IO_CACHE *info, const unsigned char *Buffer, size_t Count);
74
 
 
75
 
/**
76
 
 * @brief
77
 
 *   Lock appends for the IO_CACHE if required (need_append_buffer_lock)   
78
 
 */
79
 
inline
80
 
static void lock_append_buffer(register IO_CACHE *, int )
81
 
{
82
 
}
83
 
 
84
 
/**
85
 
 * @brief
86
 
 *   Unlock appends for the IO_CACHE if required (need_append_buffer_lock)   
87
 
 */
88
 
inline
89
 
static void unlock_append_buffer(register IO_CACHE *, int )
90
 
{
91
 
}
92
 
 
93
 
/**
94
 
 * @brief
95
 
 *   Round up to the nearest IO_SIZE boundary 
96
 
 */
97
 
inline
98
 
static size_t io_round_up(size_t x)
99
 
{
100
 
  return ((x+IO_SIZE-1) & ~(IO_SIZE-1));
101
 
}
102
 
 
103
 
/**
104
 
 * @brief
105
 
 *   Round down to the nearest IO_SIZE boundary   
106
 
 */
107
 
inline
108
 
static size_t io_round_dn(size_t x)
109
 
{
110
 
  return (x & ~(IO_SIZE-1));
111
 
}
112
 
 
113
 
 
114
 
/**
115
 
 * @brief 
116
 
 *   Setup internal pointers inside IO_CACHE
117
 
 * 
118
 
 * @details
119
 
 *   This is called on automatically on init or reinit of IO_CACHE
120
 
 *   It must be called externally if one moves or copies an IO_CACHE object.
121
 
 * 
122
 
 * @param info Cache handler to setup
123
 
 */
 
60
#define lock_append_buffer(info) \
 
61
 pthread_mutex_lock(&(info)->append_buffer_lock)
 
62
#define unlock_append_buffer(info) \
 
63
 pthread_mutex_unlock(&(info)->append_buffer_lock)
 
64
 
 
65
#define IO_ROUND_UP(X) (((X)+IO_SIZE-1) & ~(IO_SIZE-1))
 
66
#define IO_ROUND_DN(X) ( (X)            & ~(IO_SIZE-1))
 
67
 
 
68
/*
 
69
  Setup internal pointers inside IO_CACHE
 
70
 
 
71
  SYNOPSIS
 
72
    setup_io_cache()
 
73
    info                IO_CACHE handler
 
74
 
 
75
  NOTES
 
76
    This is called on automaticly on init or reinit of IO_CACHE
 
77
    It must be called externally if one moves or copies an IO_CACHE
 
78
    object.
 
79
*/
 
80
 
124
81
void setup_io_cache(IO_CACHE* info)
125
82
{
126
83
  /* Ensure that my_b_tell() and my_b_bytes_in_cache works */
151
108
      as myisamchk
152
109
    */
153
110
    break;
 
111
  case SEQ_READ_APPEND:
 
112
    info->read_function = _my_b_seq_read;
 
113
    info->write_function = 0;                   /* Force a core if used */
 
114
    break;
154
115
  default:
155
 
    info->read_function = _my_b_read;
 
116
    info->read_function =
 
117
                          info->share ? _my_b_read_r :
 
118
                                        _my_b_read;
156
119
    info->write_function = _my_b_write;
157
120
  }
158
121
 
159
122
  setup_io_cache(info);
160
123
}
161
124
 
162
 
/**
163
 
 * @brief
164
 
 *   Initialize an IO_CACHE object
165
 
 *
166
 
 * @param info Cache handler to initialize
167
 
 * @param file File that should be associated with the handler
168
 
 *                 If == -1 then real_open_cached_file() will be called when it's time to open file.
169
 
 * @param cachesize Size of buffer to allocate for read/write
170
 
 *                      If == 0 then use my_default_record_cache_size
171
 
 * @param type Type of cache
172
 
 * @param seek_offset Where cache should start reading/writing
173
 
 * @param use_async_io Set to 1 if we should use async_io (if avaiable)
174
 
 * @param cache_myflags Bitmap of different flags
175
 
                            MY_WME | MY_FAE | MY_NABP | MY_FNABP | MY_DONT_CHECK_FILESIZE
176
 
 * 
177
 
 * @retval 0 success
178
 
 * @retval # error
179
 
 */
180
 
int init_io_cache(IO_CACHE *info, int file, size_t cachesize,
 
125
 
 
126
/*
 
127
  Initialize an IO_CACHE object
 
128
 
 
129
  SYNOPSOS
 
130
    init_io_cache()
 
131
    info                cache handler to initialize
 
132
    file                File that should be associated to to the handler
 
133
                        If == -1 then real_open_cached_file()
 
134
                        will be called when it's time to open file.
 
135
    cachesize           Size of buffer to allocate for read/write
 
136
                        If == 0 then use my_default_record_cache_size
 
137
    type                Type of cache
 
138
    seek_offset         Where cache should start reading/writing
 
139
    use_async_io        Set to 1 of we should use async_io (if avaiable)
 
140
    cache_myflags       Bitmap of differnt flags
 
141
                        MY_WME | MY_FAE | MY_NABP | MY_FNABP |
 
142
                        MY_DONT_CHECK_FILESIZE
 
143
 
 
144
  RETURN
 
145
    0  ok
 
146
    #  error
 
147
*/
 
148
 
 
149
int init_io_cache(IO_CACHE *info, File file, size_t cachesize,
181
150
                  enum cache_type type, my_off_t seek_offset,
182
151
                  bool use_async_io, myf cache_myflags)
183
152
{
184
153
  size_t min_cache;
185
 
  off_t pos;
 
154
  my_off_t pos;
186
155
  my_off_t end_of_file= ~(my_off_t) 0;
187
156
 
188
157
  info->file= file;
196
165
 
197
166
  if (file >= 0)
198
167
  {
199
 
    pos= lseek(file, 0, SEEK_CUR);
200
 
    if ((pos == MY_FILEPOS_ERROR) && (errno == ESPIPE))
 
168
    pos= my_tell(file, MYF(0));
 
169
    if ((pos == (my_off_t) -1) && (my_errno == ESPIPE))
201
170
    {
202
171
      /*
203
172
         This kind of object doesn't support seek() or tell(). Don't set a
212
181
      assert(seek_offset == 0);
213
182
    }
214
183
    else
215
 
      info->seek_not_done= test(seek_offset != (my_off_t)pos);
 
184
      info->seek_not_done= test(seek_offset != pos);
216
185
  }
217
186
 
 
187
  info->disk_writes= 0;
 
188
  info->share=0;
 
189
 
218
190
  if (!cachesize && !(cachesize= my_default_record_cache_size))
219
191
    return(1);                          /* No cache requested */
220
192
  min_cache=use_async_io ? IO_SIZE*4 : IO_SIZE*2;
221
 
  if (type == READ_CACHE)
 
193
  if (type == READ_CACHE || type == SEQ_READ_APPEND)
222
194
  {                                             /* Assume file isn't growing */
223
195
    if (!(cache_myflags & MY_DONT_CHECK_FILESIZE))
224
196
    {
225
197
      /* Calculate end of file to avoid allocating oversized buffers */
226
 
      end_of_file=lseek(file,0L,SEEK_END);
 
198
      end_of_file=my_seek(file,0L,MY_SEEK_END,MYF(0));
227
199
      /* Need to reset seek_not_done now that we just did a seek. */
228
200
      info->seek_not_done= end_of_file == seek_offset ? 0 : 1;
229
201
      if (end_of_file < seek_offset)
247
219
      if (cachesize < min_cache)
248
220
        cachesize = min_cache;
249
221
      buffer_block= cachesize;
 
222
      if (type == SEQ_READ_APPEND)
 
223
        buffer_block *= 2;
250
224
      if ((info->buffer=
251
 
           (unsigned char*) malloc(buffer_block)) != 0)
 
225
           (unsigned char*) my_malloc(buffer_block,
 
226
                             MYF((cache_myflags & ~ MY_WME) |
 
227
                                 (cachesize == min_cache ? MY_WME : 0)))) != 0)
252
228
      {
253
229
        info->write_buffer=info->buffer;
254
 
        info->alloced_buffer= true;
 
230
        if (type == SEQ_READ_APPEND)
 
231
          info->write_buffer = info->buffer + cachesize;
 
232
        info->alloced_buffer=1;
255
233
        break;                                  /* Enough memory found */
256
234
      }
257
235
      if (cachesize == min_cache)
264
242
  info->read_length=info->buffer_length=cachesize;
265
243
  info->myflags=cache_myflags & ~(MY_NABP | MY_FNABP);
266
244
  info->request_pos= info->read_pos= info->write_pos = info->buffer;
 
245
  if (type == SEQ_READ_APPEND)
 
246
  {
 
247
    info->append_read_pos = info->write_pos = info->write_buffer;
 
248
    info->write_end = info->write_buffer + info->buffer_length;
 
249
    pthread_mutex_init(&info->append_buffer_lock,MY_MUTEX_INIT_FAST);
 
250
  }
 
251
#if defined(SAFE_MUTEX)
 
252
  else
 
253
  {
 
254
    /* Clear mutex so that safe_mutex will notice that it's not initialized */
 
255
    memset(&info->append_buffer_lock, 0, sizeof(info));
 
256
  }
 
257
#endif
267
258
 
268
259
  if (type == WRITE_CACHE)
269
260
    info->write_end=
313
304
}
314
305
#endif
315
306
 
316
 
/**
317
 
 * @brief 
318
 
 *   Reset the cache
319
 
 * 
320
 
 * @detail
321
 
 *   Use this to reset cache to re-start reading or to change the type 
322
 
 *   between READ_CACHE <-> WRITE_CACHE
323
 
 *   If we are doing a reinit of a cache where we have the start of the file
324
 
 *   in the cache, we are reusing this memory without flushing it to disk.
325
 
 */
 
307
 
 
308
/*
 
309
  Use this to reset cache to re-start reading or to change the type
 
310
  between READ_CACHE <-> WRITE_CACHE
 
311
  If we are doing a reinit of a cache where we have the start of the file
 
312
  in the cache, we are reusing this memory without flushing it to disk.
 
313
*/
 
314
 
326
315
bool reinit_io_cache(IO_CACHE *info, enum cache_type type,
327
316
                        my_off_t seek_offset,
328
 
                        bool use_async_io,
 
317
                        bool use_async_io __attribute__((unused)),
329
318
                        bool clear_cache)
330
319
{
331
320
  /* One can't do reinit with the following types */
332
321
  assert(type != READ_NET && info->type != READ_NET &&
333
 
              type != WRITE_NET && info->type != WRITE_NET);
 
322
              type != WRITE_NET && info->type != WRITE_NET &&
 
323
              type != SEQ_READ_APPEND && info->type != SEQ_READ_APPEND);
334
324
 
335
325
  /* If the whole file is in memory, avoid flushing to disk */
336
326
  if (! clear_cache &&
406
396
    info->read_function=_my_b_async_read;
407
397
  }
408
398
  info->inited=0;
409
 
#else
410
 
  (void)use_async_io;
411
399
#endif
412
400
  return(0);
413
401
} /* reinit_io_cache */
414
402
 
415
 
/**
416
 
 * @brief 
417
 
 *   Read buffered.
418
 
 * 
419
 
 * @detail
420
 
 *   This function is only called from the my_b_read() macro when there
421
 
 *   aren't enough characters in the buffer to satisfy the request.
422
 
 *
423
 
 * WARNING
424
 
 *   When changing this function, be careful with handling file offsets
425
 
 *   (end-of_file, pos_in_file). Do not cast them to possibly smaller
426
 
 *   types than my_off_t unless you can be sure that their value fits.
427
 
 *   Same applies to differences of file offsets.
428
 
 *
429
 
 * @param info IO_CACHE pointer @param Buffer Buffer to retrieve count bytes
430
 
 * from file @param Count Number of bytes to read into Buffer
431
 
 * 
432
 
 * @retval 0 We succeeded in reading all data
433
 
 * @retval 1 Error: can't read requested characters
434
 
 */
435
 
static int _my_b_read(register IO_CACHE *info, unsigned char *Buffer, size_t Count)
 
403
 
 
404
 
 
405
/*
 
406
  Read buffered.
 
407
 
 
408
  SYNOPSIS
 
409
    _my_b_read()
 
410
      info                      IO_CACHE pointer
 
411
      Buffer                    Buffer to retrieve count bytes from file
 
412
      Count                     Number of bytes to read into Buffer
 
413
 
 
414
  NOTE
 
415
    This function is only called from the my_b_read() macro when there
 
416
    isn't enough characters in the buffer to satisfy the request.
 
417
 
 
418
  WARNING
 
419
 
 
420
    When changing this function, be careful with handling file offsets
 
421
    (end-of_file, pos_in_file). Do not cast them to possibly smaller
 
422
    types than my_off_t unless you can be sure that their value fits.
 
423
    Same applies to differences of file offsets.
 
424
 
 
425
    When changing this function, check _my_b_read_r(). It might need the
 
426
    same change.
 
427
 
 
428
  RETURN
 
429
    0      we succeeded in reading all data
 
430
    1      Error: can't read requested characters
 
431
*/
 
432
 
 
433
int _my_b_read(register IO_CACHE *info, unsigned char *Buffer, size_t Count)
436
434
{
437
435
  size_t length,diff_length,left_length, max_length;
438
436
  my_off_t pos_in_file;
448
446
  /* pos_in_file always point on where info->buffer was read */
449
447
  pos_in_file=info->pos_in_file+ (size_t) (info->read_end - info->buffer);
450
448
 
451
 
  /*
 
449
  /* 
452
450
    Whenever a function which operates on IO_CACHE flushes/writes
453
451
    some part of the IO_CACHE to disk it will set the property
454
452
    "seek_not_done" to indicate this to other functions operating
456
454
  */
457
455
  if (info->seek_not_done)
458
456
  {
459
 
    if ((lseek(info->file,pos_in_file,SEEK_SET) != MY_FILEPOS_ERROR))
 
457
    if ((my_seek(info->file,pos_in_file,MY_SEEK_SET,MYF(0)) 
 
458
        != MY_FILEPOS_ERROR))
460
459
    {
461
460
      /* No error, reset seek_not_done flag. */
462
461
      info->seek_not_done= 0;
468
467
        info->file is a pipe or socket or FIFO.  We never should have tried
469
468
        to seek on that.  See Bugs#25807 and #22828 for more info.
470
469
      */
471
 
      assert(errno != ESPIPE);
 
470
      assert(my_errno != ESPIPE);
472
471
      info->error= -1;
473
472
      return(1);
474
473
    }
506
505
  {
507
506
    if (Count)
508
507
    {
509
 
      info->error= static_cast<int>(left_length);       /* We only got this many char */
 
508
      info->error= left_length;         /* We only got this many char */
510
509
      return(1);
511
510
    }
512
511
    length=0;                           /* Didn't read any chars */
530
529
}
531
530
 
532
531
 
 
532
/*
 
533
  Prepare IO_CACHE for shared use.
 
534
 
 
535
  SYNOPSIS
 
536
    init_io_cache_share()
 
537
      read_cache                A read cache. This will be copied for
 
538
                                every thread after setup.
 
539
      cshare                    The share.
 
540
      write_cache               If non-NULL a write cache that is to be
 
541
                                synchronized with the read caches.
 
542
      num_threads               Number of threads sharing the cache
 
543
                                including the write thread if any.
 
544
 
 
545
  DESCRIPTION
 
546
 
 
547
    The shared cache is used so: One IO_CACHE is initialized with
 
548
    init_io_cache(). This includes the allocation of a buffer. Then a
 
549
    share is allocated and init_io_cache_share() is called with the io
 
550
    cache and the share. Then the io cache is copied for each thread. So
 
551
    every thread has its own copy of IO_CACHE. But the allocated buffer
 
552
    is shared because cache->buffer is the same for all caches.
 
553
 
 
554
    One thread reads data from the file into the buffer. All threads
 
555
    read from the buffer, but every thread maintains its own set of
 
556
    pointers into the buffer. When all threads have used up the buffer
 
557
    contents, one of the threads reads the next block of data into the
 
558
    buffer. To accomplish this, each thread enters the cache lock before
 
559
    accessing the buffer. They wait in lock_io_cache() until all threads
 
560
    joined the lock. The last thread entering the lock is in charge of
 
561
    reading from file to buffer. It wakes all threads when done.
 
562
 
 
563
    Synchronizing a write cache to the read caches works so: Whenever
 
564
    the write buffer needs a flush, the write thread enters the lock and
 
565
    waits for all other threads to enter the lock too. They do this when
 
566
    they have used up the read buffer. When all threads are in the lock,
 
567
    the write thread copies the write buffer to the read buffer and
 
568
    wakes all threads.
 
569
 
 
570
    share->running_threads is the number of threads not being in the
 
571
    cache lock. When entering lock_io_cache() the number is decreased.
 
572
    When the thread that fills the buffer enters unlock_io_cache() the
 
573
    number is reset to the number of threads. The condition
 
574
    running_threads == 0 means that all threads are in the lock. Bumping
 
575
    up the number to the full count is non-intuitive. But increasing the
 
576
    number by one for each thread that leaves the lock could lead to a
 
577
    solo run of one thread. The last thread to join a lock reads from
 
578
    file to buffer, wakes the other threads, processes the data in the
 
579
    cache and enters the lock again. If no other thread left the lock
 
580
    meanwhile, it would think it's the last one again and read the next
 
581
    block...
 
582
 
 
583
    The share has copies of 'error', 'buffer', 'read_end', and
 
584
    'pos_in_file' from the thread that filled the buffer. We may not be
 
585
    able to access this information directly from its cache because the
 
586
    thread may be removed from the share before the variables could be
 
587
    copied by all other threads. Or, if a write buffer is synchronized,
 
588
    it would change its 'pos_in_file' after waking the other threads,
 
589
    possibly before they could copy its value.
 
590
 
 
591
    However, the 'buffer' variable in the share is for a synchronized
 
592
    write cache. It needs to know where to put the data. Otherwise it
 
593
    would need access to the read cache of one of the threads that is
 
594
    not yet removed from the share.
 
595
 
 
596
  RETURN
 
597
    void
 
598
*/
 
599
 
 
600
void init_io_cache_share(IO_CACHE *read_cache, IO_CACHE_SHARE *cshare,
 
601
                         IO_CACHE *write_cache, uint32_t num_threads)
 
602
{
 
603
  assert(num_threads > 1);
 
604
  assert(read_cache->type == READ_CACHE);
 
605
  assert(!write_cache || (write_cache->type == WRITE_CACHE));
 
606
 
 
607
  pthread_mutex_init(&cshare->mutex, MY_MUTEX_INIT_FAST);
 
608
  pthread_cond_init(&cshare->cond, 0);
 
609
  pthread_cond_init(&cshare->cond_writer, 0);
 
610
 
 
611
  cshare->running_threads= num_threads;
 
612
  cshare->total_threads=   num_threads;
 
613
  cshare->error=           0;    /* Initialize. */
 
614
  cshare->buffer=          read_cache->buffer;
 
615
  cshare->read_end=        NULL; /* See function comment of lock_io_cache(). */
 
616
  cshare->pos_in_file=     0;    /* See function comment of lock_io_cache(). */
 
617
  cshare->source_cache=    write_cache; /* Can be NULL. */
 
618
 
 
619
  read_cache->share=         cshare;
 
620
  read_cache->read_function= _my_b_read_r;
 
621
  read_cache->current_pos=   NULL;
 
622
  read_cache->current_end=   NULL;
 
623
 
 
624
  if (write_cache)
 
625
    write_cache->share= cshare;
 
626
 
 
627
  return;
 
628
}
 
629
 
 
630
 
 
631
/*
 
632
  Remove a thread from shared access to IO_CACHE.
 
633
 
 
634
  SYNOPSIS
 
635
    remove_io_thread()
 
636
      cache                     The IO_CACHE to be removed from the share.
 
637
 
 
638
  NOTE
 
639
 
 
640
    Every thread must do that on exit for not to deadlock other threads.
 
641
 
 
642
    The last thread destroys the pthread resources.
 
643
 
 
644
    A writer flushes its cache first.
 
645
 
 
646
  RETURN
 
647
    void
 
648
*/
 
649
 
 
650
void remove_io_thread(IO_CACHE *cache)
 
651
{
 
652
  IO_CACHE_SHARE *cshare= cache->share;
 
653
  uint32_t total;
 
654
 
 
655
  /* If the writer goes, it needs to flush the write cache. */
 
656
  if (cache == cshare->source_cache)
 
657
    flush_io_cache(cache);
 
658
 
 
659
  pthread_mutex_lock(&cshare->mutex);
 
660
 
 
661
  /* Remove from share. */
 
662
  total= --cshare->total_threads;
 
663
 
 
664
  /* Detach from share. */
 
665
  cache->share= NULL;
 
666
 
 
667
  /* If the writer goes, let the readers know. */
 
668
  if (cache == cshare->source_cache)
 
669
  {
 
670
    cshare->source_cache= NULL;
 
671
  }
 
672
 
 
673
  /* If all threads are waiting for me to join the lock, wake them. */
 
674
  if (!--cshare->running_threads)
 
675
  {
 
676
    pthread_cond_signal(&cshare->cond_writer);
 
677
    pthread_cond_broadcast(&cshare->cond);
 
678
  }
 
679
 
 
680
  pthread_mutex_unlock(&cshare->mutex);
 
681
 
 
682
  if (!total)
 
683
  {
 
684
    pthread_cond_destroy (&cshare->cond_writer);
 
685
    pthread_cond_destroy (&cshare->cond);
 
686
    pthread_mutex_destroy(&cshare->mutex);
 
687
  }
 
688
 
 
689
  return;
 
690
}
 
691
 
 
692
 
 
693
/*
 
694
  Lock IO cache and wait for all other threads to join.
 
695
 
 
696
  SYNOPSIS
 
697
    lock_io_cache()
 
698
      cache                     The cache of the thread entering the lock.
 
699
      pos                       File position of the block to read.
 
700
                                Unused for the write thread.
 
701
 
 
702
  DESCRIPTION
 
703
 
 
704
    Wait for all threads to finish with the current buffer. We want
 
705
    all threads to proceed in concert. The last thread to join
 
706
    lock_io_cache() will read the block from file and all threads start
 
707
    to use it. Then they will join again for reading the next block.
 
708
 
 
709
    The waiting threads detect a fresh buffer by comparing
 
710
    cshare->pos_in_file with the position they want to process next.
 
711
    Since the first block may start at position 0, we take
 
712
    cshare->read_end as an additional condition. This variable is
 
713
    initialized to NULL and will be set after a block of data is written
 
714
    to the buffer.
 
715
 
 
716
  RETURN
 
717
    1           OK, lock in place, go ahead and read.
 
718
    0           OK, unlocked, another thread did the read.
 
719
*/
 
720
 
 
721
static int lock_io_cache(IO_CACHE *cache, my_off_t pos)
 
722
{
 
723
  IO_CACHE_SHARE *cshare= cache->share;
 
724
 
 
725
  /* Enter the lock. */
 
726
  pthread_mutex_lock(&cshare->mutex);
 
727
  cshare->running_threads--;
 
728
 
 
729
  if (cshare->source_cache)
 
730
  {
 
731
    /* A write cache is synchronized to the read caches. */
 
732
 
 
733
    if (cache == cshare->source_cache)
 
734
    {
 
735
      /* The writer waits until all readers are here. */
 
736
      while (cshare->running_threads)
 
737
      {
 
738
        pthread_cond_wait(&cshare->cond_writer, &cshare->mutex);
 
739
      }
 
740
      /* Stay locked. Leave the lock later by unlock_io_cache(). */
 
741
      return(1);
 
742
    }
 
743
 
 
744
    /* The last thread wakes the writer. */
 
745
    if (!cshare->running_threads)
 
746
    {
 
747
      pthread_cond_signal(&cshare->cond_writer);
 
748
    }
 
749
 
 
750
    /*
 
751
      Readers wait until the data is copied from the writer. Another
 
752
      reason to stop waiting is the removal of the write thread. If this
 
753
      happens, we leave the lock with old data in the buffer.
 
754
    */
 
755
    while ((!cshare->read_end || (cshare->pos_in_file < pos)) &&
 
756
           cshare->source_cache)
 
757
    {
 
758
      pthread_cond_wait(&cshare->cond, &cshare->mutex);
 
759
    }
 
760
 
 
761
    /*
 
762
      If the writer was removed from the share while this thread was
 
763
      asleep, we need to simulate an EOF condition. The writer cannot
 
764
      reset the share variables as they might still be in use by readers
 
765
      of the last block. When we awake here then because the last
 
766
      joining thread signalled us. If the writer is not the last, it
 
767
      will not signal. So it is safe to clear the buffer here.
 
768
    */
 
769
    if (!cshare->read_end || (cshare->pos_in_file < pos))
 
770
    {
 
771
      cshare->read_end= cshare->buffer; /* Empty buffer. */
 
772
      cshare->error= 0; /* EOF is not an error. */
 
773
    }
 
774
  }
 
775
  else
 
776
  {
 
777
    /*
 
778
      There are read caches only. The last thread arriving in
 
779
      lock_io_cache() continues with a locked cache and reads the block.
 
780
    */
 
781
    if (!cshare->running_threads)
 
782
    {
 
783
      /* Stay locked. Leave the lock later by unlock_io_cache(). */
 
784
      return(1);
 
785
    }
 
786
 
 
787
    /*
 
788
      All other threads wait until the requested block is read by the
 
789
      last thread arriving. Another reason to stop waiting is the
 
790
      removal of a thread. If this leads to all threads being in the
 
791
      lock, we have to continue also. The first of the awaken threads
 
792
      will then do the read.
 
793
    */
 
794
    while ((!cshare->read_end || (cshare->pos_in_file < pos)) &&
 
795
           cshare->running_threads)
 
796
    {
 
797
      pthread_cond_wait(&cshare->cond, &cshare->mutex);
 
798
    }
 
799
 
 
800
    /* If the block is not yet read, continue with a locked cache and read. */
 
801
    if (!cshare->read_end || (cshare->pos_in_file < pos))
 
802
    {
 
803
      /* Stay locked. Leave the lock later by unlock_io_cache(). */
 
804
      return(1);
 
805
    }
 
806
 
 
807
    /* Another thread did read the block already. */
 
808
  }
 
809
 
 
810
  /*
 
811
    Leave the lock. Do not call unlock_io_cache() later. The thread that
 
812
    filled the buffer did this and marked all threads as running.
 
813
  */
 
814
  pthread_mutex_unlock(&cshare->mutex);
 
815
  return(0);
 
816
}
 
817
 
 
818
 
 
819
/*
 
820
  Unlock IO cache.
 
821
 
 
822
  SYNOPSIS
 
823
    unlock_io_cache()
 
824
      cache                     The cache of the thread leaving the lock.
 
825
 
 
826
  NOTE
 
827
    This is called by the thread that filled the buffer. It marks all
 
828
    threads as running and awakes them. This must not be done by any
 
829
    other thread.
 
830
 
 
831
    Do not signal cond_writer. Either there is no writer or the writer
 
832
    is the only one who can call this function.
 
833
 
 
834
    The reason for resetting running_threads to total_threads before
 
835
    waking all other threads is that it could be possible that this
 
836
    thread is so fast with processing the buffer that it enters the lock
 
837
    before even one other thread has left it. If every awoken thread
 
838
    would increase running_threads by one, this thread could think that
 
839
    he is again the last to join and would not wait for the other
 
840
    threads to process the data.
 
841
 
 
842
  RETURN
 
843
    void
 
844
*/
 
845
 
 
846
static void unlock_io_cache(IO_CACHE *cache)
 
847
{
 
848
  IO_CACHE_SHARE *cshare= cache->share;
 
849
 
 
850
  cshare->running_threads= cshare->total_threads;
 
851
  pthread_cond_broadcast(&cshare->cond);
 
852
  pthread_mutex_unlock(&cshare->mutex);
 
853
  return;
 
854
}
 
855
 
 
856
 
 
857
/*
 
858
  Read from IO_CACHE when it is shared between several threads.
 
859
 
 
860
  SYNOPSIS
 
861
    _my_b_read_r()
 
862
      cache                     IO_CACHE pointer
 
863
      Buffer                    Buffer to retrieve count bytes from file
 
864
      Count                     Number of bytes to read into Buffer
 
865
 
 
866
  NOTE
 
867
    This function is only called from the my_b_read() macro when there
 
868
    isn't enough characters in the buffer to satisfy the request.
 
869
 
 
870
  IMPLEMENTATION
 
871
 
 
872
    It works as follows: when a thread tries to read from a file (that
 
873
    is, after using all the data from the (shared) buffer), it just
 
874
    hangs on lock_io_cache(), waiting for other threads. When the very
 
875
    last thread attempts a read, lock_io_cache() returns 1, the thread
 
876
    does actual IO and unlock_io_cache(), which signals all the waiting
 
877
    threads that data is in the buffer.
 
878
 
 
879
  WARNING
 
880
 
 
881
    When changing this function, be careful with handling file offsets
 
882
    (end-of_file, pos_in_file). Do not cast them to possibly smaller
 
883
    types than my_off_t unless you can be sure that their value fits.
 
884
    Same applies to differences of file offsets. (Bug #11527)
 
885
 
 
886
    When changing this function, check _my_b_read(). It might need the
 
887
    same change.
 
888
 
 
889
  RETURN
 
890
    0      we succeeded in reading all data
 
891
    1      Error: can't read requested characters
 
892
*/
 
893
 
 
894
int _my_b_read_r(register IO_CACHE *cache, unsigned char *Buffer, size_t Count)
 
895
{
 
896
  my_off_t pos_in_file;
 
897
  size_t length, diff_length, left_length;
 
898
  IO_CACHE_SHARE *cshare= cache->share;
 
899
 
 
900
  if ((left_length= (size_t) (cache->read_end - cache->read_pos)))
 
901
  {
 
902
    assert(Count >= left_length);       /* User is not using my_b_read() */
 
903
    memcpy(Buffer, cache->read_pos, left_length);
 
904
    Buffer+= left_length;
 
905
    Count-= left_length;
 
906
  }
 
907
  while (Count)
 
908
  {
 
909
    size_t cnt, len;
 
910
 
 
911
    pos_in_file= cache->pos_in_file + (cache->read_end - cache->buffer);
 
912
    diff_length= (size_t) (pos_in_file & (IO_SIZE-1));
 
913
    length=IO_ROUND_UP(Count+diff_length)-diff_length;
 
914
    length= ((length <= cache->read_length) ?
 
915
             length + IO_ROUND_DN(cache->read_length - length) :
 
916
             length - IO_ROUND_UP(length - cache->read_length));
 
917
    if (cache->type != READ_FIFO &&
 
918
        (length > (cache->end_of_file - pos_in_file)))
 
919
      length= (size_t) (cache->end_of_file - pos_in_file);
 
920
    if (length == 0)
 
921
    {
 
922
      cache->error= (int) left_length;
 
923
      return(1);
 
924
    }
 
925
    if (lock_io_cache(cache, pos_in_file))
 
926
    {
 
927
      /* With a synchronized write/read cache we won't come here... */
 
928
      assert(!cshare->source_cache);
 
929
      /*
 
930
        ... unless the writer has gone before this thread entered the
 
931
        lock. Simulate EOF in this case. It can be distinguished by
 
932
        cache->file.
 
933
      */
 
934
      if (cache->file < 0)
 
935
        len= 0;
 
936
      else
 
937
      {
 
938
        /*
 
939
          Whenever a function which operates on IO_CACHE flushes/writes
 
940
          some part of the IO_CACHE to disk it will set the property
 
941
          "seek_not_done" to indicate this to other functions operating
 
942
          on the IO_CACHE.
 
943
        */
 
944
        if (cache->seek_not_done)
 
945
        {
 
946
          if (my_seek(cache->file, pos_in_file, MY_SEEK_SET, MYF(0))
 
947
              == MY_FILEPOS_ERROR)
 
948
          {
 
949
            cache->error= -1;
 
950
            unlock_io_cache(cache);
 
951
            return(1);
 
952
          }
 
953
        }
 
954
        len= my_read(cache->file, cache->buffer, length, cache->myflags);
 
955
      }
 
956
      cache->read_end=    cache->buffer + (len == (size_t) -1 ? 0 : len);
 
957
      cache->error=       (len == length ? 0 : (int) len);
 
958
      cache->pos_in_file= pos_in_file;
 
959
 
 
960
      /* Copy important values to the share. */
 
961
      cshare->error=       cache->error;
 
962
      cshare->read_end=    cache->read_end;
 
963
      cshare->pos_in_file= pos_in_file;
 
964
 
 
965
      /* Mark all threads as running and wake them. */
 
966
      unlock_io_cache(cache);
 
967
    }
 
968
    else
 
969
    {
 
970
      /*
 
971
        With a synchronized write/read cache readers always come here.
 
972
        Copy important values from the share.
 
973
      */
 
974
      cache->error=       cshare->error;
 
975
      cache->read_end=    cshare->read_end;
 
976
      cache->pos_in_file= cshare->pos_in_file;
 
977
 
 
978
      len= ((cache->error == -1) ? (size_t) -1 :
 
979
            (size_t) (cache->read_end - cache->buffer));
 
980
    }
 
981
    cache->read_pos=      cache->buffer;
 
982
    cache->seek_not_done= 0;
 
983
    if (len == 0 || len == (size_t) -1)
 
984
    {
 
985
      cache->error= (int) left_length;
 
986
      return(1);
 
987
    }
 
988
    cnt= (len > Count) ? Count : len;
 
989
    memcpy(Buffer, cache->read_pos, cnt);
 
990
    Count -= cnt;
 
991
    Buffer+= cnt;
 
992
    left_length+= cnt;
 
993
    cache->read_pos+= cnt;
 
994
  }
 
995
  return(0);
 
996
}
 
997
 
 
998
 
 
999
/*
 
1000
  Copy data from write cache to read cache.
 
1001
 
 
1002
  SYNOPSIS
 
1003
    copy_to_read_buffer()
 
1004
      write_cache               The write cache.
 
1005
      write_buffer              The source of data, mostly the cache buffer.
 
1006
      write_length              The number of bytes to copy.
 
1007
 
 
1008
  NOTE
 
1009
    The write thread will wait for all read threads to join the cache
 
1010
    lock. Then it copies the data over and wakes the read threads.
 
1011
 
 
1012
  RETURN
 
1013
    void
 
1014
*/
 
1015
 
 
1016
static void copy_to_read_buffer(IO_CACHE *write_cache,
 
1017
                                const unsigned char *write_buffer, size_t write_length)
 
1018
{
 
1019
  IO_CACHE_SHARE *cshare= write_cache->share;
 
1020
 
 
1021
  assert(cshare->source_cache == write_cache);
 
1022
  /*
 
1023
    write_length is usually less or equal to buffer_length.
 
1024
    It can be bigger if _my_b_write() is called with a big length.
 
1025
  */
 
1026
  while (write_length)
 
1027
  {
 
1028
    size_t copy_length= cmin(write_length, write_cache->buffer_length);
 
1029
    int  __attribute__((unused)) rc;
 
1030
 
 
1031
    rc= lock_io_cache(write_cache, write_cache->pos_in_file);
 
1032
    /* The writing thread does always have the lock when it awakes. */
 
1033
    assert(rc);
 
1034
 
 
1035
    memcpy(cshare->buffer, write_buffer, copy_length);
 
1036
 
 
1037
    cshare->error=       0;
 
1038
    cshare->read_end=    cshare->buffer + copy_length;
 
1039
    cshare->pos_in_file= write_cache->pos_in_file;
 
1040
 
 
1041
    /* Mark all threads as running and wake them. */
 
1042
    unlock_io_cache(write_cache);
 
1043
 
 
1044
    write_buffer+= copy_length;
 
1045
    write_length-= copy_length;
 
1046
  }
 
1047
}
 
1048
 
 
1049
 
 
1050
/*
 
1051
  Do sequential read from the SEQ_READ_APPEND cache.
 
1052
  
 
1053
  We do this in three stages:
 
1054
   - first read from info->buffer
 
1055
   - then if there are still data to read, try the file descriptor
 
1056
   - afterwards, if there are still data to read, try append buffer
 
1057
 
 
1058
  RETURNS
 
1059
    0  Success
 
1060
    1  Failed to read
 
1061
*/
 
1062
 
 
1063
int _my_b_seq_read(register IO_CACHE *info, unsigned char *Buffer, size_t Count)
 
1064
{
 
1065
  size_t length, diff_length, left_length, save_count, max_length;
 
1066
  my_off_t pos_in_file;
 
1067
  save_count=Count;
 
1068
 
 
1069
  /* first, read the regular buffer */
 
1070
  if ((left_length=(size_t) (info->read_end-info->read_pos)))
 
1071
  {
 
1072
    assert(Count > left_length);        /* User is not using my_b_read() */
 
1073
    memcpy(Buffer,info->read_pos, left_length);
 
1074
    Buffer+=left_length;
 
1075
    Count-=left_length;
 
1076
  }
 
1077
  lock_append_buffer(info);
 
1078
 
 
1079
  /* pos_in_file always point on where info->buffer was read */
 
1080
  if ((pos_in_file=info->pos_in_file +
 
1081
       (size_t) (info->read_end - info->buffer)) >= info->end_of_file)
 
1082
    goto read_append_buffer;
 
1083
 
 
1084
  /*
 
1085
    With read-append cache we must always do a seek before we read,
 
1086
    because the write could have moved the file pointer astray
 
1087
  */
 
1088
  if (my_seek(info->file,pos_in_file,MY_SEEK_SET,MYF(0)) == MY_FILEPOS_ERROR)
 
1089
  {
 
1090
   info->error= -1;
 
1091
   unlock_append_buffer(info);
 
1092
   return (1);
 
1093
  }
 
1094
  info->seek_not_done=0;
 
1095
 
 
1096
  diff_length= (size_t) (pos_in_file & (IO_SIZE-1));
 
1097
 
 
1098
  /* now the second stage begins - read from file descriptor */
 
1099
  if (Count >= (size_t) (IO_SIZE+(IO_SIZE-diff_length)))
 
1100
  {
 
1101
    /* Fill first intern buffer */
 
1102
    size_t read_length;
 
1103
 
 
1104
    length=(Count & (size_t) ~(IO_SIZE-1))-diff_length;
 
1105
    if ((read_length= my_read(info->file,Buffer, length,
 
1106
                              info->myflags)) == (size_t) -1)
 
1107
    {
 
1108
      info->error= -1;
 
1109
      unlock_append_buffer(info);
 
1110
      return 1;
 
1111
    }
 
1112
    Count-=read_length;
 
1113
    Buffer+=read_length;
 
1114
    pos_in_file+=read_length;
 
1115
 
 
1116
    if (read_length != length)
 
1117
    {
 
1118
      /*
 
1119
        We only got part of data;  Read the rest of the data from the
 
1120
        write buffer
 
1121
      */
 
1122
      goto read_append_buffer;
 
1123
    }
 
1124
    left_length+=length;
 
1125
    diff_length=0;
 
1126
  }
 
1127
 
 
1128
  max_length= info->read_length-diff_length;
 
1129
  if (max_length > (info->end_of_file - pos_in_file))
 
1130
    max_length= (size_t) (info->end_of_file - pos_in_file);
 
1131
  if (!max_length)
 
1132
  {
 
1133
    if (Count)
 
1134
      goto read_append_buffer;
 
1135
    length=0;                           /* Didn't read any more chars */
 
1136
  }
 
1137
  else
 
1138
  {
 
1139
    length= my_read(info->file,info->buffer, max_length, info->myflags);
 
1140
    if (length == (size_t) -1)
 
1141
    {
 
1142
      info->error= -1;
 
1143
      unlock_append_buffer(info);
 
1144
      return 1;
 
1145
    }
 
1146
    if (length < Count)
 
1147
    {
 
1148
      memcpy(Buffer, info->buffer, length);
 
1149
      Count -= length;
 
1150
      Buffer += length;
 
1151
 
 
1152
      /*
 
1153
         added the line below to make
 
1154
         assert(pos_in_file==info->end_of_file) pass.
 
1155
         otherwise this does not appear to be needed
 
1156
      */
 
1157
      pos_in_file += length;
 
1158
      goto read_append_buffer;
 
1159
    }
 
1160
  }
 
1161
  unlock_append_buffer(info);
 
1162
  info->read_pos=info->buffer+Count;
 
1163
  info->read_end=info->buffer+length;
 
1164
  info->pos_in_file=pos_in_file;
 
1165
  memcpy(Buffer,info->buffer,(size_t) Count);
 
1166
  return 0;
 
1167
 
 
1168
read_append_buffer:
 
1169
 
 
1170
  /*
 
1171
     Read data from the current write buffer.
 
1172
     Count should never be == 0 here (The code will work even if count is 0)
 
1173
  */
 
1174
 
 
1175
  {
 
1176
    /* First copy the data to Count */
 
1177
    size_t len_in_buff = (size_t) (info->write_pos - info->append_read_pos);
 
1178
    size_t copy_len;
 
1179
    size_t transfer_len;
 
1180
 
 
1181
    assert(info->append_read_pos <= info->write_pos);
 
1182
    /*
 
1183
      TODO: figure out if the assert below is needed or correct.
 
1184
    */
 
1185
    assert(pos_in_file == info->end_of_file);
 
1186
    copy_len=cmin(Count, len_in_buff);
 
1187
    memcpy(Buffer, info->append_read_pos, copy_len);
 
1188
    info->append_read_pos += copy_len;
 
1189
    Count -= copy_len;
 
1190
    if (Count)
 
1191
      info->error = save_count - Count;
 
1192
 
 
1193
    /* Fill read buffer with data from write buffer */
 
1194
    memcpy(info->buffer, info->append_read_pos,
 
1195
           (size_t) (transfer_len=len_in_buff - copy_len));
 
1196
    info->read_pos= info->buffer;
 
1197
    info->read_end= info->buffer+transfer_len;
 
1198
    info->append_read_pos=info->write_pos;
 
1199
    info->pos_in_file=pos_in_file+copy_len;
 
1200
    info->end_of_file+=len_in_buff;
 
1201
  }
 
1202
  unlock_append_buffer(info);
 
1203
  return Count ? 1 : 0;
 
1204
}
 
1205
 
 
1206
 
533
1207
#ifdef HAVE_AIOWAIT
534
1208
 
535
 
/**
536
 
 * @brief
537
 
 *   Read from the IO_CACHE into a buffer and feed asynchronously from disk when needed.
538
 
 *
539
 
 * @param info IO_CACHE pointer
540
 
 * @param Buffer Buffer to retrieve count bytes from file
541
 
 * @param Count Number of bytes to read into Buffer
542
 
 * 
543
 
 * @retval -1 An error has occurred; errno is set.
544
 
 * @retval 0 Success
545
 
 * @retval 1 An error has occurred; IO_CACHE to error state.
546
 
 */
 
1209
/*
 
1210
  Read from the IO_CACHE into a buffer and feed asynchronously
 
1211
  from disk when needed.
 
1212
 
 
1213
  SYNOPSIS
 
1214
    _my_b_async_read()
 
1215
      info                      IO_CACHE pointer
 
1216
      Buffer                    Buffer to retrieve count bytes from file
 
1217
      Count                     Number of bytes to read into Buffer
 
1218
 
 
1219
  RETURN VALUE
 
1220
    -1          An error has occurred; my_errno is set.
 
1221
     0          Success
 
1222
     1          An error has occurred; IO_CACHE to error state.
 
1223
*/
 
1224
 
547
1225
int _my_b_async_read(register IO_CACHE *info, unsigned char *Buffer, size_t Count)
548
1226
{
549
1227
  size_t length,read_length,diff_length,left_length,use_length,org_Count;
567
1245
        my_error(EE_READ, MYF(ME_BELL+ME_WAITTANG),
568
1246
                 my_filename(info->file),
569
1247
                 info->aio_result.result.aio_errno);
570
 
      errno=info->aio_result.result.aio_errno;
 
1248
      my_errno=info->aio_result.result.aio_errno;
571
1249
      info->error= -1;
572
1250
      return(1);
573
1251
    }
574
1252
    if (! (read_length= (size_t) info->aio_result.result.aio_return) ||
575
1253
        read_length == (size_t) -1)
576
1254
    {
577
 
      errno=0;                          /* For testing */
 
1255
      my_errno=0;                               /* For testing */
578
1256
      info->error= (read_length == (size_t) -1 ? -1 :
579
1257
                    (int) (read_length+left_length));
580
1258
      return(1);
607
1285
      }
608
1286
    }
609
1287
        /* Copy found bytes to buffer */
610
 
    length=min(Count,read_length);
 
1288
    length=cmin(Count,read_length);
611
1289
    memcpy(Buffer,info->read_pos,(size_t) length);
612
1290
    Buffer+=length;
613
1291
    Count-=length;
627
1305
      info->error=(int) (read_length+left_length);
628
1306
      return 1;
629
1307
    }
630
 
 
631
 
    if (lseek(info->file,next_pos_in_file,SEEK_SET) == MY_FILEPOS_ERROR)
 
1308
    
 
1309
    if (my_seek(info->file,next_pos_in_file,MY_SEEK_SET,MYF(0))
 
1310
        == MY_FILEPOS_ERROR)
632
1311
    {
633
1312
      info->error= -1;
634
1313
      return (1);
640
1319
      if ((read_length=my_read(info->file,info->request_pos,
641
1320
                               read_length, info->myflags)) == (size_t) -1)
642
1321
        return info->error= -1;
643
 
      use_length=min(Count,read_length);
 
1322
      use_length=cmin(Count,read_length);
644
1323
      memcpy(Buffer,info->request_pos,(size_t) use_length);
645
1324
      info->read_pos=info->request_pos+Count;
646
1325
      info->read_end=info->request_pos+read_length;
651
1330
      {                                 /* Didn't find hole block */
652
1331
        if (info->myflags & (MY_WME | MY_FAE | MY_FNABP) && Count != org_Count)
653
1332
          my_error(EE_EOFERR, MYF(ME_BELL+ME_WAITTANG),
654
 
                   my_filename(info->file),errno);
 
1333
                   my_filename(info->file),my_errno);
655
1334
        info->error=(int) (read_length+left_length);
656
1335
        return 1;
657
1336
      }
684
1363
  {
685
1364
    info->aio_result.result.aio_errno=AIO_INPROGRESS;   /* Marker for test */
686
1365
    if (aioread(info->file,read_buffer, max_length,
687
 
                (my_off_t) next_pos_in_file,SEEK_SET,
 
1366
                (my_off_t) next_pos_in_file,MY_SEEK_SET,
688
1367
                &info->aio_result.result))
689
1368
    {                                           /* Skip async io */
690
 
      errno=errno;
 
1369
      my_errno=errno;
691
1370
      if (info->request_pos != info->buffer)
692
1371
      {
693
 
        memmove(info->buffer, info->request_pos,
694
 
                (size_t) (info->read_end - info->read_pos));
 
1372
        memcpy(info->buffer, info->request_pos,
 
1373
               (size_t) (info->read_end - info->read_pos));
695
1374
        info->request_pos=info->buffer;
696
1375
        info->read_pos-=info->read_length;
697
1376
        info->read_end-=info->read_length;
707
1386
#endif
708
1387
 
709
1388
 
710
 
/**
711
 
 * @brief
712
 
 *   Read one byte when buffer is empty
713
 
 */
 
1389
/* Read one byte when buffer is empty */
 
1390
 
714
1391
int _my_b_get(IO_CACHE *info)
715
1392
{
716
1393
  unsigned char buff;
724
1401
  return (int) (unsigned char) buff;
725
1402
}
726
1403
 
727
 
/**
728
 
 * @brief
729
 
 *   Write a byte buffer to IO_CACHE and flush to disk if IO_CACHE is full.
730
 
 *
731
 
 * @retval -1 On error; errno contains error code.
732
 
 * @retval 0 On success
733
 
 * @retval 1 On error on write
734
 
 */
 
1404
/* 
 
1405
   Write a byte buffer to IO_CACHE and flush to disk
 
1406
   if IO_CACHE is full.
 
1407
 
 
1408
   RETURN VALUE
 
1409
    1 On error on write
 
1410
    0 On success
 
1411
   -1 On error; my_errno contains error code.
 
1412
*/
 
1413
 
735
1414
int _my_b_write(register IO_CACHE *info, const unsigned char *Buffer, size_t Count)
736
1415
{
737
1416
  size_t rest_length,length;
738
1417
 
739
1418
  if (info->pos_in_file+info->buffer_length > info->end_of_file)
740
1419
  {
741
 
    errno=errno=EFBIG;
 
1420
    my_errno=errno=EFBIG;
742
1421
    return info->error = -1;
743
1422
  }
744
1423
 
761
1440
        "seek_not_done" to indicate this to other functions operating
762
1441
        on the IO_CACHE.
763
1442
      */
764
 
      if (lseek(info->file,info->pos_in_file,SEEK_SET))
 
1443
      if (my_seek(info->file,info->pos_in_file,MY_SEEK_SET,MYF(0)))
765
1444
      {
766
1445
        info->error= -1;
767
1446
        return (1);
771
1450
    if (my_write(info->file, Buffer, length, info->myflags | MY_NABP))
772
1451
      return info->error= -1;
773
1452
 
 
1453
    /*
 
1454
      In case of a shared I/O cache with a writer we normally do direct
 
1455
      write cache to read cache copy. Simulate this here by direct
 
1456
      caller buffer to read cache copy. Do it after the write so that
 
1457
      the cache readers actions on the flushed part can go in parallel
 
1458
      with the write of the extra stuff. copy_to_read_buffer()
 
1459
      synchronizes writer and readers so that after this call the
 
1460
      readers can act on the extra stuff while the writer can go ahead
 
1461
      and prepare the next output. copy_to_read_buffer() relies on
 
1462
      info->pos_in_file.
 
1463
    */
 
1464
    if (info->share)
 
1465
      copy_to_read_buffer(info, Buffer, length);
 
1466
 
774
1467
    Count-=length;
775
1468
    Buffer+=length;
776
1469
    info->pos_in_file+=length;
780
1473
  return 0;
781
1474
}
782
1475
 
783
 
/**
784
 
 * @brief
785
 
 *   Write a block to disk where part of the data may be inside the record buffer.
786
 
 *   As all write calls to the data goes through the cache,
787
 
 *   we will never get a seek over the end of the buffer.
788
 
 */
 
1476
 
 
1477
/*
 
1478
  Append a block to the write buffer.
 
1479
  This is done with the buffer locked to ensure that we don't read from
 
1480
  the write buffer before we are ready with it.
 
1481
*/
 
1482
 
 
1483
int my_b_append(register IO_CACHE *info, const unsigned char *Buffer, size_t Count)
 
1484
{
 
1485
  size_t rest_length,length;
 
1486
 
 
1487
  /*
 
1488
    Assert that we cannot come here with a shared cache. If we do one
 
1489
    day, we might need to add a call to copy_to_read_buffer().
 
1490
  */
 
1491
  assert(!info->share);
 
1492
 
 
1493
  lock_append_buffer(info);
 
1494
  rest_length= (size_t) (info->write_end - info->write_pos);
 
1495
  if (Count <= rest_length)
 
1496
    goto end;
 
1497
  memcpy(info->write_pos, Buffer, rest_length);
 
1498
  Buffer+=rest_length;
 
1499
  Count-=rest_length;
 
1500
  info->write_pos+=rest_length;
 
1501
  if (my_b_flush_io_cache(info,0))
 
1502
  {
 
1503
    unlock_append_buffer(info);
 
1504
    return 1;
 
1505
  }
 
1506
  if (Count >= IO_SIZE)
 
1507
  {                                     /* Fill first intern buffer */
 
1508
    length=Count & (size_t) ~(IO_SIZE-1);
 
1509
    if (my_write(info->file,Buffer, length, info->myflags | MY_NABP))
 
1510
    {
 
1511
      unlock_append_buffer(info);
 
1512
      return info->error= -1;
 
1513
    }
 
1514
    Count-=length;
 
1515
    Buffer+=length;
 
1516
    info->end_of_file+=length;
 
1517
  }
 
1518
 
 
1519
end:
 
1520
  memcpy(info->write_pos,Buffer,(size_t) Count);
 
1521
  info->write_pos+=Count;
 
1522
  unlock_append_buffer(info);
 
1523
  return 0;
 
1524
}
 
1525
 
 
1526
 
 
1527
int my_b_safe_write(IO_CACHE *info, const unsigned char *Buffer, size_t Count)
 
1528
{
 
1529
  /*
 
1530
    Sasha: We are not writing this with the ? operator to avoid hitting
 
1531
    a possible compiler bug. At least gcc 2.95 cannot deal with 
 
1532
    several layers of ternary operators that evaluated comma(,) operator
 
1533
    expressions inside - I do have a test case if somebody wants it
 
1534
  */
 
1535
  if (info->type == SEQ_READ_APPEND)
 
1536
    return my_b_append(info, Buffer, Count);
 
1537
  return my_b_write(info, Buffer, Count);
 
1538
}
 
1539
 
 
1540
 
 
1541
/*
 
1542
  Write a block to disk where part of the data may be inside the record
 
1543
  buffer.  As all write calls to the data goes through the cache,
 
1544
  we will never get a seek over the end of the buffer
 
1545
*/
 
1546
 
789
1547
int my_block_write(register IO_CACHE *info, const unsigned char *Buffer, size_t Count,
790
1548
                   my_off_t pos)
791
1549
{
792
1550
  size_t length;
793
1551
  int error=0;
794
1552
 
 
1553
  /*
 
1554
    Assert that we cannot come here with a shared cache. If we do one
 
1555
    day, we might need to add a call to copy_to_read_buffer().
 
1556
  */
 
1557
  assert(!info->share);
 
1558
 
795
1559
  if (pos < info->pos_in_file)
796
1560
  {
797
1561
    /* Of no overlap, write everything without buffering */
798
1562
    if (pos + Count <= info->pos_in_file)
799
1563
      return (pwrite(info->file, Buffer, Count, pos) == 0);
800
1564
    /* Write the part of the block that is before buffer */
801
 
    length= (uint32_t) (info->pos_in_file - pos);
 
1565
    length= (uint) (info->pos_in_file - pos);
802
1566
    if (pwrite(info->file, Buffer, length, pos) == 0)
803
1567
      info->error= error= -1;
804
1568
    Buffer+=length;
805
1569
    pos+=  length;
806
1570
    Count-= length;
 
1571
#ifndef HAVE_PREAD
 
1572
    info->seek_not_done=1;
 
1573
#endif
807
1574
  }
808
1575
 
809
1576
  /* Check if we want to write inside the used part of the buffer.*/
829
1596
  return error;
830
1597
}
831
1598
 
832
 
/**
833
 
 * @brief
834
 
 *   Flush write cache 
835
 
 */
 
1599
 
 
1600
        /* Flush write cache */
 
1601
 
 
1602
#define LOCK_APPEND_BUFFER if (need_append_buffer_lock) \
 
1603
  lock_append_buffer(info);
 
1604
#define UNLOCK_APPEND_BUFFER if (need_append_buffer_lock) \
 
1605
  unlock_append_buffer(info);
 
1606
 
836
1607
int my_b_flush_io_cache(IO_CACHE *info, int need_append_buffer_lock)
837
1608
{
838
1609
  size_t length;
839
 
  bool append_cache= false;
 
1610
  bool append_cache;
840
1611
  my_off_t pos_in_file;
841
1612
 
 
1613
  if (!(append_cache = (info->type == SEQ_READ_APPEND)))
 
1614
    need_append_buffer_lock=0;
 
1615
 
842
1616
  if (info->type == WRITE_CACHE || append_cache)
843
1617
  {
844
1618
    if (info->file == -1)
846
1620
      if (real_open_cached_file(info))
847
1621
        return((info->error= -1));
848
1622
    }
849
 
    lock_append_buffer(info, need_append_buffer_lock);
 
1623
    LOCK_APPEND_BUFFER;
850
1624
 
851
1625
    if ((length=(size_t) (info->write_pos - info->write_buffer)))
852
1626
    {
 
1627
      /*
 
1628
        In case of a shared I/O cache with a writer we do direct write
 
1629
        cache to read cache copy. Do it before the write here so that
 
1630
        the readers can work in parallel with the write.
 
1631
        copy_to_read_buffer() relies on info->pos_in_file.
 
1632
      */
 
1633
      if (info->share)
 
1634
        copy_to_read_buffer(info, info->write_buffer, length);
 
1635
 
853
1636
      pos_in_file=info->pos_in_file;
854
1637
      /*
855
1638
        If we have append cache, we always open the file with
857
1640
      */
858
1641
      if (!append_cache && info->seek_not_done)
859
1642
      {                                 /* File touched, do seek */
860
 
        if (lseek(info->file,pos_in_file,SEEK_SET) == MY_FILEPOS_ERROR)
 
1643
        if (my_seek(info->file,pos_in_file,MY_SEEK_SET,MYF(0)) ==
 
1644
            MY_FILEPOS_ERROR)
861
1645
        {
862
 
          unlock_append_buffer(info, need_append_buffer_lock);
 
1646
          UNLOCK_APPEND_BUFFER;
863
1647
          return((info->error= -1));
864
1648
        }
865
1649
        if (!append_cache)
882
1666
      else
883
1667
      {
884
1668
        info->end_of_file+=(info->write_pos-info->append_read_pos);
885
 
        my_off_t tell_ret= lseek(info->file, 0, SEEK_CUR);
886
 
        assert(info->end_of_file == tell_ret);
 
1669
        assert(info->end_of_file == my_tell(info->file,MYF(0)));
887
1670
      }
888
1671
 
889
1672
      info->append_read_pos=info->write_pos=info->write_buffer;
890
 
      unlock_append_buffer(info, need_append_buffer_lock);
 
1673
      ++info->disk_writes;
 
1674
      UNLOCK_APPEND_BUFFER;
891
1675
      return(info->error);
892
1676
    }
893
1677
  }
898
1682
    info->inited=0;
899
1683
  }
900
1684
#endif
901
 
  unlock_append_buffer(info, need_append_buffer_lock);
 
1685
  UNLOCK_APPEND_BUFFER;
902
1686
  return(0);
903
1687
}
904
1688
 
905
 
/**
906
 
 * @brief
907
 
 *   Free an IO_CACHE object
908
 
 * 
909
 
 * @detail
910
 
 *   It's currently safe to call this if one has called init_io_cache()
911
 
 *   on the 'info' object, even if init_io_cache() failed.
912
 
 *   This function is also safe to call twice with the same handle.
913
 
 * 
914
 
 * @param info IO_CACHE Handle to free
915
 
 * 
916
 
 * @retval 0 ok
917
 
 * @retval # Error
918
 
 */
 
1689
/*
 
1690
  Free an IO_CACHE object
 
1691
 
 
1692
  SYNOPSOS
 
1693
    end_io_cache()
 
1694
    info                IO_CACHE Handle to free
 
1695
 
 
1696
  NOTES
 
1697
    It's currently safe to call this if one has called init_io_cache()
 
1698
    on the 'info' object, even if init_io_cache() failed.
 
1699
    This function is also safe to call twice with the same handle.
 
1700
 
 
1701
  RETURN
 
1702
   0  ok
 
1703
   #  Error
 
1704
*/
 
1705
 
919
1706
int end_io_cache(IO_CACHE *info)
920
1707
{
921
1708
  int error=0;
922
1709
  IO_CACHE_CALLBACK pre_close;
923
1710
 
 
1711
  /*
 
1712
    Every thread must call remove_io_thread(). The last one destroys
 
1713
    the share elements.
 
1714
  */
 
1715
  assert(!info->share || !info->share->total_threads);
 
1716
 
924
1717
  if ((pre_close=info->pre_close))
925
1718
  {
926
1719
    (*pre_close)(info);
934
1727
    free((unsigned char*) info->buffer);
935
1728
    info->buffer=info->read_pos=(unsigned char*) 0;
936
1729
  }
937
 
 
 
1730
  if (info->type == SEQ_READ_APPEND)
 
1731
  {
 
1732
    /* Destroy allocated mutex */
 
1733
    info->type= TYPE_NOT_SET;
 
1734
    pthread_mutex_destroy(&info->append_buffer_lock);
 
1735
  }
938
1736
  return(error);
939
1737
} /* end_io_cache */
940
1738
 
941
 
} /* namespace internal */
942
 
} /* namespace drizzled */
 
1739
 
 
1740
/**********************************************************************
 
1741
 Testing of MF_IOCACHE
 
1742
**********************************************************************/
 
1743
 
 
1744
#ifdef MAIN
 
1745
 
 
1746
#include <my_dir.h>
 
1747
 
 
1748
void die(const char* fmt, ...)
 
1749
{
 
1750
  va_list va_args;
 
1751
  va_start(va_args,fmt);
 
1752
  fprintf(stderr,"Error:");
 
1753
  vfprintf(stderr, fmt,va_args);
 
1754
  fprintf(stderr,", errno=%d\n", errno);
 
1755
  exit(1);
 
1756
}
 
1757
 
 
1758
int open_file(const char* fname, IO_CACHE* info, int cache_size)
 
1759
{
 
1760
  int fd;
 
1761
  if ((fd=my_open(fname,O_CREAT | O_RDWR,MYF(MY_WME))) < 0)
 
1762
    die("Could not open %s", fname);
 
1763
  if (init_io_cache(info, fd, cache_size, SEQ_READ_APPEND, 0,0,MYF(MY_WME)))
 
1764
    die("failed in init_io_cache()");
 
1765
  return fd;
 
1766
}
 
1767
 
 
1768
void close_file(IO_CACHE* info)
 
1769
{
 
1770
  end_io_cache(info);
 
1771
  my_close(info->file, MYF(MY_WME));
 
1772
}
 
1773
 
 
1774
int main(int argc, char** argv)
 
1775
{
 
1776
  IO_CACHE sra_cache; /* SEQ_READ_APPEND */
 
1777
  MY_STAT status;
 
1778
  const char* fname="/tmp/iocache.test";
 
1779
  int cache_size=16384;
 
1780
  char llstr_buf[22];
 
1781
  int max_block,total_bytes=0;
 
1782
  int i,num_loops=100,error=0;
 
1783
  char *p;
 
1784
  char* block, *block_end;
 
1785
  MY_INIT(argv[0]);
 
1786
  max_block = cache_size*3;
 
1787
  if (!(block=(char*)my_malloc(max_block,MYF(MY_WME))))
 
1788
    die("Not enough memory to allocate test block");
 
1789
  block_end = block + max_block;
 
1790
  for (p = block,i=0; p < block_end;i++)
 
1791
  {
 
1792
    *p++ = (char)i;
 
1793
  }
 
1794
  if (my_stat(fname,&status, MYF(0)) &&
 
1795
      my_delete(fname,MYF(MY_WME)))
 
1796
    {
 
1797
      die("Delete of %s failed, aborting", fname);
 
1798
    }
 
1799
  open_file(fname,&sra_cache, cache_size);
 
1800
  for (i = 0; i < num_loops; i++)
 
1801
  {
 
1802
    char buf[4];
 
1803
    int block_size = abs(rand() % max_block);
 
1804
    int4store(buf, block_size);
 
1805
    if (my_b_append(&sra_cache,buf,4) ||
 
1806
        my_b_append(&sra_cache, block, block_size))
 
1807
      die("write failed");
 
1808
    total_bytes += 4+block_size;
 
1809
  }
 
1810
  close_file(&sra_cache);
 
1811
  free(block);
 
1812
  if (!my_stat(fname,&status,MYF(MY_WME)))
 
1813
    die("%s failed to stat, but I had just closed it,\
 
1814
 wonder how that happened");
 
1815
  printf("Final size of %s is %s, wrote %d bytes\n",fname,
 
1816
         llstr(status.st_size,llstr_buf),
 
1817
         total_bytes);
 
1818
  my_delete(fname, MYF(MY_WME));
 
1819
  /* check correctness of tests */
 
1820
  if (total_bytes != status.st_size)
 
1821
  {
 
1822
    fprintf(stderr,"Not the same number of bytes acutally  in file as bytes \
 
1823
supposedly written\n");
 
1824
    error=1;
 
1825
  }
 
1826
  exit(error);
 
1827
  return 0;
 
1828
}
 
1829
#endif