541
Prepare IO_CACHE for shared use.
544
init_io_cache_share()
545
read_cache A read cache. This will be copied for
546
every thread after setup.
548
write_cache If non-NULL a write cache that is to be
549
synchronized with the read caches.
550
num_threads Number of threads sharing the cache
551
including the write thread if any.
555
The shared cache is used so: One IO_CACHE is initialized with
556
init_io_cache(). This includes the allocation of a buffer. Then a
557
share is allocated and init_io_cache_share() is called with the io
558
cache and the share. Then the io cache is copied for each thread. So
559
every thread has its own copy of IO_CACHE. But the allocated buffer
560
is shared because cache->buffer is the same for all caches.
562
One thread reads data from the file into the buffer. All threads
563
read from the buffer, but every thread maintains its own set of
564
pointers into the buffer. When all threads have used up the buffer
565
contents, one of the threads reads the next block of data into the
566
buffer. To accomplish this, each thread enters the cache lock before
567
accessing the buffer. They wait in lock_io_cache() until all threads
568
joined the lock. The last thread entering the lock is in charge of
569
reading from file to buffer. It wakes all threads when done.
571
Synchronizing a write cache to the read caches works so: Whenever
572
the write buffer needs a flush, the write thread enters the lock and
573
waits for all other threads to enter the lock too. They do this when
574
they have used up the read buffer. When all threads are in the lock,
575
the write thread copies the write buffer to the read buffer and
578
share->running_threads is the number of threads not being in the
579
cache lock. When entering lock_io_cache() the number is decreased.
580
When the thread that fills the buffer enters unlock_io_cache() the
581
number is reset to the number of threads. The condition
582
running_threads == 0 means that all threads are in the lock. Bumping
583
up the number to the full count is non-intuitive. But increasing the
584
number by one for each thread that leaves the lock could lead to a
585
solo run of one thread. The last thread to join a lock reads from
586
file to buffer, wakes the other threads, processes the data in the
587
cache and enters the lock again. If no other thread left the lock
588
meanwhile, it would think it's the last one again and read the next
591
The share has copies of 'error', 'buffer', 'read_end', and
592
'pos_in_file' from the thread that filled the buffer. We may not be
593
able to access this information directly from its cache because the
594
thread may be removed from the share before the variables could be
595
copied by all other threads. Or, if a write buffer is synchronized,
596
it would change its 'pos_in_file' after waking the other threads,
597
possibly before they could copy its value.
599
However, the 'buffer' variable in the share is for a synchronized
600
write cache. It needs to know where to put the data. Otherwise it
601
would need access to the read cache of one of the threads that is
602
not yet removed from the share.
608
void init_io_cache_share(IO_CACHE *read_cache, IO_CACHE_SHARE *cshare,
609
IO_CACHE *write_cache, uint32_t num_threads)
611
assert(num_threads > 1);
612
assert(read_cache->type == READ_CACHE);
613
assert(!write_cache || (write_cache->type == WRITE_CACHE));
615
pthread_mutex_init(&cshare->mutex, MY_MUTEX_INIT_FAST);
616
pthread_cond_init(&cshare->cond, 0);
617
pthread_cond_init(&cshare->cond_writer, 0);
619
cshare->running_threads= num_threads;
620
cshare->total_threads= num_threads;
621
cshare->error= 0; /* Initialize. */
622
cshare->buffer= read_cache->buffer;
623
cshare->read_end= NULL; /* See function comment of lock_io_cache(). */
624
cshare->pos_in_file= 0; /* See function comment of lock_io_cache(). */
625
cshare->source_cache= write_cache; /* Can be NULL. */
627
read_cache->share= cshare;
628
read_cache->read_function= _my_b_read_r;
629
read_cache->current_pos= NULL;
630
read_cache->current_end= NULL;
633
write_cache->share= cshare;
640
Remove a thread from shared access to IO_CACHE.
644
cache The IO_CACHE to be removed from the share.
648
Every thread must do that on exit for not to deadlock other threads.
650
The last thread destroys the pthread resources.
652
A writer flushes its cache first.
658
void remove_io_thread(IO_CACHE *cache)
660
IO_CACHE_SHARE *cshare= cache->share;
663
/* If the writer goes, it needs to flush the write cache. */
664
if (cache == cshare->source_cache)
665
flush_io_cache(cache);
667
pthread_mutex_lock(&cshare->mutex);
669
/* Remove from share. */
670
total= --cshare->total_threads;
672
/* Detach from share. */
675
/* If the writer goes, let the readers know. */
676
if (cache == cshare->source_cache)
678
cshare->source_cache= NULL;
681
/* If all threads are waiting for me to join the lock, wake them. */
682
if (!--cshare->running_threads)
684
pthread_cond_signal(&cshare->cond_writer);
685
pthread_cond_broadcast(&cshare->cond);
688
pthread_mutex_unlock(&cshare->mutex);
692
pthread_cond_destroy (&cshare->cond_writer);
693
pthread_cond_destroy (&cshare->cond);
694
pthread_mutex_destroy(&cshare->mutex);
702
Lock IO cache and wait for all other threads to join.
706
cache The cache of the thread entering the lock.
707
pos File position of the block to read.
708
Unused for the write thread.
712
Wait for all threads to finish with the current buffer. We want
713
all threads to proceed in concert. The last thread to join
714
lock_io_cache() will read the block from file and all threads start
715
to use it. Then they will join again for reading the next block.
717
The waiting threads detect a fresh buffer by comparing
718
cshare->pos_in_file with the position they want to process next.
719
Since the first block may start at position 0, we take
720
cshare->read_end as an additional condition. This variable is
721
initialized to NULL and will be set after a block of data is written
725
1 OK, lock in place, go ahead and read.
726
0 OK, unlocked, another thread did the read.
729
static int lock_io_cache(IO_CACHE *cache, my_off_t pos)
731
IO_CACHE_SHARE *cshare= cache->share;
733
/* Enter the lock. */
734
pthread_mutex_lock(&cshare->mutex);
735
cshare->running_threads--;
737
if (cshare->source_cache)
739
/* A write cache is synchronized to the read caches. */
741
if (cache == cshare->source_cache)
743
/* The writer waits until all readers are here. */
744
while (cshare->running_threads)
746
pthread_cond_wait(&cshare->cond_writer, &cshare->mutex);
748
/* Stay locked. Leave the lock later by unlock_io_cache(). */
752
/* The last thread wakes the writer. */
753
if (!cshare->running_threads)
755
pthread_cond_signal(&cshare->cond_writer);
759
Readers wait until the data is copied from the writer. Another
760
reason to stop waiting is the removal of the write thread. If this
761
happens, we leave the lock with old data in the buffer.
763
while ((!cshare->read_end || (cshare->pos_in_file < pos)) &&
764
cshare->source_cache)
766
pthread_cond_wait(&cshare->cond, &cshare->mutex);
770
If the writer was removed from the share while this thread was
771
asleep, we need to simulate an EOF condition. The writer cannot
772
reset the share variables as they might still be in use by readers
773
of the last block. When we awake here then because the last
774
joining thread signalled us. If the writer is not the last, it
775
will not signal. So it is safe to clear the buffer here.
777
if (!cshare->read_end || (cshare->pos_in_file < pos))
779
cshare->read_end= cshare->buffer; /* Empty buffer. */
780
cshare->error= 0; /* EOF is not an error. */
786
There are read caches only. The last thread arriving in
787
lock_io_cache() continues with a locked cache and reads the block.
789
if (!cshare->running_threads)
791
/* Stay locked. Leave the lock later by unlock_io_cache(). */
796
All other threads wait until the requested block is read by the
797
last thread arriving. Another reason to stop waiting is the
798
removal of a thread. If this leads to all threads being in the
799
lock, we have to continue also. The first of the awaken threads
800
will then do the read.
802
while ((!cshare->read_end || (cshare->pos_in_file < pos)) &&
803
cshare->running_threads)
805
pthread_cond_wait(&cshare->cond, &cshare->mutex);
808
/* If the block is not yet read, continue with a locked cache and read. */
809
if (!cshare->read_end || (cshare->pos_in_file < pos))
811
/* Stay locked. Leave the lock later by unlock_io_cache(). */
815
/* Another thread did read the block already. */
819
Leave the lock. Do not call unlock_io_cache() later. The thread that
820
filled the buffer did this and marked all threads as running.
822
pthread_mutex_unlock(&cshare->mutex);
832
cache The cache of the thread leaving the lock.
835
This is called by the thread that filled the buffer. It marks all
836
threads as running and awakes them. This must not be done by any
839
Do not signal cond_writer. Either there is no writer or the writer
840
is the only one who can call this function.
842
The reason for resetting running_threads to total_threads before
843
waking all other threads is that it could be possible that this
844
thread is so fast with processing the buffer that it enters the lock
845
before even one other thread has left it. If every awoken thread
846
would increase running_threads by one, this thread could think that
847
he is again the last to join and would not wait for the other
848
threads to process the data.
854
static void unlock_io_cache(IO_CACHE *cache)
856
IO_CACHE_SHARE *cshare= cache->share;
858
cshare->running_threads= cshare->total_threads;
859
pthread_cond_broadcast(&cshare->cond);
860
pthread_mutex_unlock(&cshare->mutex);
866
Read from IO_CACHE when it is shared between several threads.
870
cache IO_CACHE pointer
871
Buffer Buffer to retrieve count bytes from file
872
Count Number of bytes to read into Buffer
875
This function is only called from the my_b_read() macro when there
876
isn't enough characters in the buffer to satisfy the request.
880
It works as follows: when a thread tries to read from a file (that
881
is, after using all the data from the (shared) buffer), it just
882
hangs on lock_io_cache(), waiting for other threads. When the very
883
last thread attempts a read, lock_io_cache() returns 1, the thread
884
does actual IO and unlock_io_cache(), which signals all the waiting
885
threads that data is in the buffer.
889
When changing this function, be careful with handling file offsets
890
(end-of_file, pos_in_file). Do not cast them to possibly smaller
891
types than my_off_t unless you can be sure that their value fits.
892
Same applies to differences of file offsets. (Bug #11527)
894
When changing this function, check _my_b_read(). It might need the
898
0 we succeeded in reading all data
899
1 Error: can't read requested characters
902
int _my_b_read_r(register IO_CACHE *cache, unsigned char *Buffer, size_t Count)
904
my_off_t pos_in_file;
905
size_t length, diff_length, left_length;
906
IO_CACHE_SHARE *cshare= cache->share;
908
if ((left_length= (size_t) (cache->read_end - cache->read_pos)))
910
assert(Count >= left_length); /* User is not using my_b_read() */
911
memcpy(Buffer, cache->read_pos, left_length);
912
Buffer+= left_length;
919
pos_in_file= cache->pos_in_file + (cache->read_end - cache->buffer);
920
diff_length= (size_t) (pos_in_file & (IO_SIZE-1));
921
length=IO_ROUND_UP(Count+diff_length)-diff_length;
922
length= ((length <= cache->read_length) ?
923
length + IO_ROUND_DN(cache->read_length - length) :
924
length - IO_ROUND_UP(length - cache->read_length));
925
if (cache->type != READ_FIFO &&
926
(length > (cache->end_of_file - pos_in_file)))
927
length= (size_t) (cache->end_of_file - pos_in_file);
930
cache->error= (int) left_length;
933
if (lock_io_cache(cache, pos_in_file))
935
/* With a synchronized write/read cache we won't come here... */
936
assert(!cshare->source_cache);
938
... unless the writer has gone before this thread entered the
939
lock. Simulate EOF in this case. It can be distinguished by
947
Whenever a function which operates on IO_CACHE flushes/writes
948
some part of the IO_CACHE to disk it will set the property
949
"seek_not_done" to indicate this to other functions operating
952
if (cache->seek_not_done)
954
if (lseek(cache->file, pos_in_file, SEEK_SET) == MY_FILEPOS_ERROR)
957
unlock_io_cache(cache);
961
len= my_read(cache->file, cache->buffer, length, cache->myflags);
963
cache->read_end= cache->buffer + (len == (size_t) -1 ? 0 : len);
964
cache->error= (len == length ? 0 : (int) len);
965
cache->pos_in_file= pos_in_file;
967
/* Copy important values to the share. */
968
cshare->error= cache->error;
969
cshare->read_end= cache->read_end;
970
cshare->pos_in_file= pos_in_file;
972
/* Mark all threads as running and wake them. */
973
unlock_io_cache(cache);
978
With a synchronized write/read cache readers always come here.
979
Copy important values from the share.
981
cache->error= cshare->error;
982
cache->read_end= cshare->read_end;
983
cache->pos_in_file= cshare->pos_in_file;
985
len= ((cache->error == -1) ? (size_t) -1 :
986
(size_t) (cache->read_end - cache->buffer));
988
cache->read_pos= cache->buffer;
989
cache->seek_not_done= 0;
990
if (len == 0 || len == (size_t) -1)
992
cache->error= (int) left_length;
995
cnt= (len > Count) ? Count : len;
996
memcpy(Buffer, cache->read_pos, cnt);
1000
cache->read_pos+= cnt;
1007
Copy data from write cache to read cache.
1010
copy_to_read_buffer()
1011
write_cache The write cache.
1012
write_buffer The source of data, mostly the cache buffer.
1013
write_length The number of bytes to copy.
1016
The write thread will wait for all read threads to join the cache
1017
lock. Then it copies the data over and wakes the read threads.
1023
static void copy_to_read_buffer(IO_CACHE *write_cache,
1024
const unsigned char *write_buffer, size_t write_length)
1026
IO_CACHE_SHARE *cshare= write_cache->share;
1028
assert(cshare->source_cache == write_cache);
1030
write_length is usually less or equal to buffer_length.
1031
It can be bigger if _my_b_write() is called with a big length.
1033
while (write_length)
1035
size_t copy_length= min(write_length, write_cache->buffer_length);
1038
rc= lock_io_cache(write_cache, write_cache->pos_in_file);
1039
/* The writing thread does always have the lock when it awakes. */
1042
memcpy(cshare->buffer, write_buffer, copy_length);
1045
cshare->read_end= cshare->buffer + copy_length;
1046
cshare->pos_in_file= write_cache->pos_in_file;
1048
/* Mark all threads as running and wake them. */
1049
unlock_io_cache(write_cache);
1051
write_buffer+= copy_length;
1052
write_length-= copy_length;
1058
Do sequential read from the SEQ_READ_APPEND cache.
1060
We do this in three stages:
1061
- first read from info->buffer
1062
- then if there are still data to read, try the file descriptor
1063
- afterwards, if there are still data to read, try append buffer
1070
int _my_b_seq_read(register IO_CACHE *info, unsigned char *Buffer, size_t Count)
1072
size_t length, diff_length, left_length, save_count, max_length;
1073
my_off_t pos_in_file;
1076
/* first, read the regular buffer */
1077
if ((left_length=(size_t) (info->read_end-info->read_pos)))
1079
assert(Count > left_length); /* User is not using my_b_read() */
1080
memcpy(Buffer,info->read_pos, left_length);
1081
Buffer+=left_length;
1084
lock_append_buffer(info);
1086
/* pos_in_file always point on where info->buffer was read */
1087
if ((pos_in_file=info->pos_in_file +
1088
(size_t) (info->read_end - info->buffer)) >= info->end_of_file)
1089
goto read_append_buffer;
1092
With read-append cache we must always do a seek before we read,
1093
because the write could have moved the file pointer astray
1095
if (lseek(info->file,pos_in_file,SEEK_SET) == MY_FILEPOS_ERROR)
1098
unlock_append_buffer(info);
1101
info->seek_not_done=0;
1103
diff_length= (size_t) (pos_in_file & (IO_SIZE-1));
1105
/* now the second stage begins - read from file descriptor */
1106
if (Count >= (size_t) (IO_SIZE+(IO_SIZE-diff_length)))
1108
/* Fill first intern buffer */
1111
length=(Count & (size_t) ~(IO_SIZE-1))-diff_length;
1112
if ((read_length= my_read(info->file,Buffer, length,
1113
info->myflags)) == (size_t) -1)
1116
unlock_append_buffer(info);
1120
Buffer+=read_length;
1121
pos_in_file+=read_length;
1123
if (read_length != length)
1126
We only got part of data; Read the rest of the data from the
1129
goto read_append_buffer;
1131
left_length+=length;
1135
max_length= info->read_length-diff_length;
1136
if (max_length > (info->end_of_file - pos_in_file))
1137
max_length= (size_t) (info->end_of_file - pos_in_file);
1141
goto read_append_buffer;
1142
length=0; /* Didn't read any more chars */
1146
length= my_read(info->file,info->buffer, max_length, info->myflags);
1147
if (length == (size_t) -1)
1150
unlock_append_buffer(info);
1155
memcpy(Buffer, info->buffer, length);
1160
added the line below to make
1161
assert(pos_in_file==info->end_of_file) pass.
1162
otherwise this does not appear to be needed
1164
pos_in_file += length;
1165
goto read_append_buffer;
1168
unlock_append_buffer(info);
1169
info->read_pos=info->buffer+Count;
1170
info->read_end=info->buffer+length;
1171
info->pos_in_file=pos_in_file;
1172
memcpy(Buffer,info->buffer,(size_t) Count);
1178
Read data from the current write buffer.
1179
Count should never be == 0 here (The code will work even if count is 0)
1183
/* First copy the data to Count */
1184
size_t len_in_buff = (size_t) (info->write_pos - info->append_read_pos);
1186
size_t transfer_len;
1188
assert(info->append_read_pos <= info->write_pos);
1190
TODO: figure out if the assert below is needed or correct.
1192
assert(pos_in_file == info->end_of_file);
1193
copy_len=min(Count, len_in_buff);
1194
memcpy(Buffer, info->append_read_pos, copy_len);
1195
info->append_read_pos += copy_len;
1198
info->error = static_cast<int>(save_count - Count);
1200
/* Fill read buffer with data from write buffer */
1201
memcpy(info->buffer, info->append_read_pos,
1202
(size_t) (transfer_len=len_in_buff - copy_len));
1203
info->read_pos= info->buffer;
1204
info->read_end= info->buffer+transfer_len;
1205
info->append_read_pos=info->write_pos;
1206
info->pos_in_file=pos_in_file+copy_len;
1207
info->end_of_file+=len_in_buff;
1209
unlock_append_buffer(info);
1210
return Count ? 1 : 0;
1214
533
#ifdef HAVE_AIOWAIT
1217
Read from the IO_CACHE into a buffer and feed asynchronously
1218
from disk when needed.
1222
info IO_CACHE pointer
1223
Buffer Buffer to retrieve count bytes from file
1224
Count Number of bytes to read into Buffer
1227
-1 An error has occurred; errno is set.
1229
1 An error has occurred; IO_CACHE to error state.
537
* Read from the IO_CACHE into a buffer and feed asynchronously from disk when needed.
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
543
* @retval -1 An error has occurred; errno is set.
545
* @retval 1 An error has occurred; IO_CACHE to error state.
1232
547
int _my_b_async_read(register IO_CACHE *info, unsigned char *Buffer, size_t Count)
1234
549
size_t length,read_length,diff_length,left_length,use_length,org_Count;
added
removed
drizzled/internal/mf_iocache.cc
