1 /*
2 Copyright (c) 2000, 2011, Oracle and/or its affiliates
3 Copyright (c) 2010, 2020, MariaDB
4
5 This program is free software; you can redistribute it and/or modify
6 it under the terms of the GNU General Public License as published by
7 the Free Software Foundation; version 2 of the License.
8
9 This program is distributed in the hope that it will be useful,
10 but WITHOUT ANY WARRANTY; without even the implied warranty of
11 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 GNU General Public License for more details.
13
14 You should have received a copy of the GNU General Public License
15 along with this program; if not, write to the Free Software
16 Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1335 USA */
17
18 /*
19 Cashing of files with only does (sequential) read or writes of fixed-
20 length records. A read isn't allowed to go over file-length. A read is ok
21 if it ends at file-length and next read can try to read after file-length
22 (and get a EOF-error).
23 Possibly use of asyncronic io.
24 macros for read and writes for faster io.
25 Used instead of FILE when reading or writing whole files.
26 One can change info->pos_in_file to a higher value to skip bytes in file if
27 also info->read_pos is set to info->read_end.
28 If called through open_cached_file(), then the temporary file will
29 only be created if a write exeeds the file buffer or if one calls
30 my_b_flush_io_cache().
31
32 If one uses SEQ_READ_APPEND, then two buffers are allocated, one for
33 reading and another for writing. Reads are first done from disk and
34 then done from the write buffer. This is an efficient way to read
35 from a log file when one is writing to it at the same time.
36 For this to work, the file has to be opened in append mode!
37 Note that when one uses SEQ_READ_APPEND, one MUST write using
38 my_b_append ! This is needed because we need to lock the mutex
39 every time we access the write buffer.
40
41 TODO:
42 When one SEQ_READ_APPEND and we are reading and writing at the same time,
43 each time the write buffer gets full and it's written to disk, we will
44 always do a disk read to read a part of the buffer from disk to the
45 read buffer.
46 This should be fixed so that when we do a my_b_flush_io_cache() and
47 we have been reading the write buffer, we should transfer the rest of the
48 write buffer to the read buffer before we start to reuse it.
49 */
50
51 #include "mysys_priv.h"
52 #include <m_string.h>
53 #ifdef HAVE_AIOWAIT
54 #include "mysys_err.h"
55 static void my_aiowait(my_aio_result *result);
56 #endif
57 #include <errno.h>
58
59 #define lock_append_buffer(info) \
60 mysql_mutex_lock(&(info)->append_buffer_lock)
61 #define unlock_append_buffer(info) \
62 mysql_mutex_unlock(&(info)->append_buffer_lock)
63
64 #define IO_ROUND_UP(X) (((X)+IO_SIZE-1) & ~(IO_SIZE-1))
65 #define IO_ROUND_DN(X) ( (X) & ~(IO_SIZE-1))
66
67 static int _my_b_cache_read(IO_CACHE *info, uchar *Buffer, size_t Count);
68 static int _my_b_cache_read_r(IO_CACHE *info, uchar *Buffer, size_t Count);
69 static int _my_b_seq_read(IO_CACHE *info, uchar *Buffer, size_t Count);
70 static int _my_b_cache_write(IO_CACHE *info, const uchar *Buffer, size_t Count);
71 static int _my_b_cache_write_r(IO_CACHE *info, const uchar *Buffer, size_t Count);
72
73 int (*_my_b_encr_read)(IO_CACHE *info,uchar *Buffer,size_t Count)= 0;
74 int (*_my_b_encr_write)(IO_CACHE *info,const uchar *Buffer,size_t Count)= 0;
75
76
77
78 static void
init_functions(IO_CACHE * info)79 init_functions(IO_CACHE* info)
80 {
81 enum cache_type type= info->type;
82 info->read_function = 0; /* Force a core if used */
83 info->write_function = 0; /* Force a core if used */
84 switch (type) {
85 case READ_NET:
86 /*
87 Must be initialized by the caller. The problem is that
88 _my_b_net_read has to be defined in sql directory because of
89 the dependency on THD, and therefore cannot be visible to
90 programs that link against mysys but know nothing about THD, such
91 as myisamchk
92 */
93 DBUG_ASSERT(!(info->myflags & MY_ENCRYPT));
94 break;
95 case SEQ_READ_APPEND:
96 info->read_function = _my_b_seq_read;
97 DBUG_ASSERT(!(info->myflags & MY_ENCRYPT));
98 break;
99 case READ_CACHE:
100 if (info->myflags & MY_ENCRYPT)
101 {
102 DBUG_ASSERT(info->share == 0);
103 info->read_function = _my_b_encr_read;
104 break;
105 }
106 /* fall through */
107 case WRITE_CACHE:
108 if (info->myflags & MY_ENCRYPT)
109 {
110 info->write_function = _my_b_encr_write;
111 break;
112 }
113 /* fall through */
114 case READ_FIFO:
115 DBUG_ASSERT(!(info->myflags & MY_ENCRYPT));
116 info->read_function = info->share ? _my_b_cache_read_r : _my_b_cache_read;
117 info->write_function = info->share ? _my_b_cache_write_r : _my_b_cache_write;
118 break;
119 case TYPE_NOT_SET:
120 DBUG_ASSERT(0);
121 break;
122 }
123 if (type == READ_CACHE || type == WRITE_CACHE || type == SEQ_READ_APPEND)
124 info->myflags|= MY_FULL_IO;
125 else
126 info->myflags&= ~MY_FULL_IO;
127 }
128
129
130 /*
131 Initialize an IO_CACHE object
132
133 SYNOPSOS
134 init_io_cache()
135 info cache handler to initialize
136 file File that should be associated to to the handler
137 If == -1 then real_open_cached_file()
138 will be called when it's time to open file.
139 cachesize Size of buffer to allocate for read/write
140 If == 0 then use my_default_record_cache_size
141 type Type of cache
142 seek_offset Where cache should start reading/writing
143 use_async_io Set to 1 of we should use async_io (if available)
144 cache_myflags Bitmap of different flags
145 MY_WME | MY_FAE | MY_NABP | MY_FNABP |
146 MY_DONT_CHECK_FILESIZE
147
148 RETURN
149 0 ok
150 # error
151 */
152
init_io_cache(IO_CACHE * info,File file,size_t cachesize,enum cache_type type,my_off_t seek_offset,my_bool use_async_io,myf cache_myflags)153 int init_io_cache(IO_CACHE *info, File file, size_t cachesize,
154 enum cache_type type, my_off_t seek_offset,
155 my_bool use_async_io, myf cache_myflags)
156 {
157 size_t min_cache;
158 my_off_t pos;
159 my_off_t end_of_file= ~(my_off_t) 0;
160 DBUG_ENTER("init_io_cache");
161 DBUG_PRINT("enter",("cache:%p type: %d pos: %llu",
162 info, (int) type, (ulonglong) seek_offset));
163
164 info->file= file;
165 info->type= TYPE_NOT_SET; /* Don't set it until mutex are created */
166 info->pos_in_file= seek_offset;
167 info->alloced_buffer = 0;
168 info->buffer=0;
169 info->seek_not_done= 0;
170 info->next_file_user= NULL;
171
172 if (file >= 0)
173 {
174 DBUG_ASSERT(!(cache_myflags & MY_ENCRYPT));
175 pos= mysql_file_tell(file, MYF(0));
176 if ((pos == (my_off_t) -1) && (my_errno == ESPIPE))
177 {
178 /*
179 This kind of object doesn't support seek() or tell(). Don't set a
180 seek_not_done that will make us again try to seek() later and fail.
181
182 Additionally, if we're supposed to start somewhere other than the
183 the beginning of whatever this file is, then somebody made a bad
184 assumption.
185 */
186 DBUG_ASSERT(seek_offset == 0);
187 }
188 else
189 info->seek_not_done= MY_TEST(seek_offset != pos);
190 }
191 else
192 if (type == WRITE_CACHE && _my_b_encr_read)
193 {
194 cache_myflags|= MY_ENCRYPT;
195 DBUG_ASSERT(seek_offset == 0);
196 }
197
198 info->disk_writes= 0;
199 info->share=0;
200
201 if (!cachesize && !(cachesize= my_default_record_cache_size))
202 DBUG_RETURN(1); /* No cache requested */
203 min_cache=use_async_io ? IO_SIZE*4 : IO_SIZE*2;
204 if (type == READ_CACHE || type == SEQ_READ_APPEND)
205 { /* Assume file isn't growing */
206 DBUG_ASSERT(!(cache_myflags & MY_ENCRYPT));
207 if (!(cache_myflags & MY_DONT_CHECK_FILESIZE))
208 {
209 /* Calculate end of file to avoid allocating oversized buffers */
210 end_of_file= mysql_file_seek(file, 0L, MY_SEEK_END, MYF(0));
211 /* Need to reset seek_not_done now that we just did a seek. */
212 info->seek_not_done= end_of_file == seek_offset ? 0 : 1;
213 if (end_of_file < seek_offset)
214 end_of_file=seek_offset;
215 /* Trim cache size if the file is very small */
216 if ((my_off_t) cachesize > end_of_file-seek_offset+IO_SIZE*2-1)
217 {
218 cachesize= (size_t) (end_of_file-seek_offset)+IO_SIZE*2-1;
219 use_async_io=0; /* No need to use async */
220 }
221 }
222 }
223 cache_myflags &= ~MY_DONT_CHECK_FILESIZE;
224 if (type != READ_NET)
225 {
226 /* Retry allocating memory in smaller blocks until we get one */
227 cachesize= ((cachesize + min_cache-1) & ~(min_cache-1));
228 for (;;)
229 {
230 size_t buffer_block;
231 /*
232 Unset MY_WAIT_IF_FULL bit if it is set, to prevent conflict with
233 MY_ZEROFILL.
234 */
235 myf flags= (myf) (cache_myflags & ~(MY_WME | MY_WAIT_IF_FULL));
236
237 if (cachesize < min_cache)
238 cachesize = min_cache;
239 buffer_block= cachesize;
240 if (type == SEQ_READ_APPEND)
241 buffer_block *= 2;
242 else if (cache_myflags & MY_ENCRYPT)
243 buffer_block= 2*(buffer_block + MY_AES_BLOCK_SIZE) + sizeof(IO_CACHE_CRYPT);
244 if (cachesize == min_cache)
245 flags|= (myf) MY_WME;
246
247 if ((info->buffer= (uchar*) my_malloc(buffer_block, flags)) != 0)
248 {
249 if (type == SEQ_READ_APPEND)
250 info->write_buffer= info->buffer + cachesize;
251 else
252 info->write_buffer= info->buffer;
253 info->alloced_buffer= buffer_block;
254 break; /* Enough memory found */
255 }
256 if (cachesize == min_cache)
257 DBUG_RETURN(2); /* Can't alloc cache */
258 /* Try with less memory */
259 cachesize= (cachesize*3/4 & ~(min_cache-1));
260 }
261 }
262
263 DBUG_PRINT("info",("init_io_cache: cachesize = %lu", (ulong) cachesize));
264 info->read_length=info->buffer_length=cachesize;
265 info->myflags=cache_myflags & ~(MY_NABP | MY_FNABP);
266 info->request_pos= info->read_pos= info->write_pos = info->buffer;
267 if (type == SEQ_READ_APPEND)
268 {
269 info->append_read_pos = info->write_pos = info->write_buffer;
270 info->write_end = info->write_buffer + info->buffer_length;
271 mysql_mutex_init(key_IO_CACHE_append_buffer_lock,
272 &info->append_buffer_lock, MY_MUTEX_INIT_FAST);
273 }
274 #if defined(SAFE_MUTEX)
275 else
276 {
277 /* Clear mutex so that safe_mutex will notice that it's not initialized */
278 bzero((char*) &info->append_buffer_lock, sizeof(info->append_buffer_lock));
279 }
280 #endif
281
282 if (type == WRITE_CACHE)
283 info->write_end=
284 info->buffer+info->buffer_length- (seek_offset & (IO_SIZE-1));
285 else
286 info->read_end=info->buffer; /* Nothing in cache */
287
288 /* End_of_file may be changed by user later */
289 info->end_of_file= end_of_file;
290 info->error=0;
291 info->type= type;
292 init_functions(info);
293 #ifdef HAVE_AIOWAIT
294 if (use_async_io && ! my_disable_async_io)
295 {
296 DBUG_PRINT("info",("Using async io"));
297 DBUG_ASSERT(!(cache_myflags & MY_ENCRYPT));
298 info->read_length/=2;
299 info->read_function=_my_b_async_read;
300 }
301 info->inited=info->aio_result.pending=0;
302 #endif
303 DBUG_RETURN(0);
304 } /* init_io_cache */
305
306
307
308 /*
309 Initialize the slave IO_CACHE to read the same file (and data)
310 as master does.
311
312 One can create multiple slaves from a single master. Every slave and master
313 will have independent file positions.
314
315 The master must be a non-shared READ_CACHE.
316 It is assumed that no more reads are done after a master and/or a slave
317 has been freed (this limitation can be easily lifted).
318 */
319
init_slave_io_cache(IO_CACHE * master,IO_CACHE * slave)320 int init_slave_io_cache(IO_CACHE *master, IO_CACHE *slave)
321 {
322 uchar *slave_buf;
323 DBUG_ASSERT(master->type == READ_CACHE);
324 DBUG_ASSERT(!master->share);
325 DBUG_ASSERT(master->alloced_buffer);
326
327 if (!(slave_buf= (uchar*)my_malloc(master->alloced_buffer, MYF(0))))
328 {
329 return 1;
330 }
331 memcpy(slave, master, sizeof(IO_CACHE));
332 slave->buffer= slave_buf;
333
334 memcpy(slave->buffer, master->buffer, master->alloced_buffer);
335 slave->read_pos= slave->buffer + (master->read_pos - master->buffer);
336 slave->read_end= slave->buffer + (master->read_end - master->buffer);
337
338 if (master->next_file_user)
339 {
340 IO_CACHE *p;
341 for (p= master->next_file_user;
342 p->next_file_user !=master;
343 p= p->next_file_user)
344 {}
345
346 p->next_file_user= slave;
347 slave->next_file_user= master;
348 }
349 else
350 {
351 slave->next_file_user= master;
352 master->next_file_user= slave;
353 }
354 return 0;
355 }
356
357
end_slave_io_cache(IO_CACHE * cache)358 void end_slave_io_cache(IO_CACHE *cache)
359 {
360 /* Remove the cache from the next_file_user circular linked list. */
361 if (cache->next_file_user != cache)
362 {
363 IO_CACHE *p= cache->next_file_user;
364 while (p->next_file_user != cache)
365 p= p->next_file_user;
366 p->next_file_user= cache->next_file_user;
367
368 }
369 my_free(cache->buffer);
370 }
371
372 /*
373 Seek a read io cache to a given offset
374 */
seek_io_cache(IO_CACHE * cache,my_off_t needed_offset)375 void seek_io_cache(IO_CACHE *cache, my_off_t needed_offset)
376 {
377 my_off_t cached_data_start= cache->pos_in_file;
378 my_off_t cached_data_end= cache->pos_in_file + (cache->read_end -
379 cache->buffer);
380
381 if (needed_offset >= cached_data_start &&
382 needed_offset < cached_data_end)
383 {
384 /*
385 The offset we're seeking to is in the buffer.
386 Move buffer's read position accordingly
387 */
388 cache->read_pos= cache->buffer + (needed_offset - cached_data_start);
389 }
390 else
391 {
392 if (needed_offset > cache->end_of_file)
393 needed_offset= cache->end_of_file;
394 /*
395 The offset we're seeking to is not in the buffer.
396 - Set the buffer to be exhausted.
397 - Make the next read to a mysql_file_seek() call to the required
398 offset.
399 TODO(cvicentiu, spetrunia) properly implement aligned seeks for
400 efficiency.
401 */
402 cache->seek_not_done= 1;
403 cache->pos_in_file= needed_offset;
404 /* When reading it must appear as if we've started from the offset
405 that we've seeked here. We must let _my_b_cache_read assume that
406 by implying "no reading starting from pos_in_file" has happened. */
407 cache->read_pos= cache->buffer;
408 cache->read_end= cache->buffer;
409 }
410 }
411
412 /* Wait until current request is ready */
413
414 #ifdef HAVE_AIOWAIT
my_aiowait(my_aio_result * result)415 static void my_aiowait(my_aio_result *result)
416 {
417 if (result->pending)
418 {
419 struct aio_result_t *tmp;
420 for (;;)
421 {
422 if ((int) (tmp=aiowait((struct timeval *) 0)) == -1)
423 {
424 if (errno == EINTR)
425 continue;
426 DBUG_PRINT("error",("No aio request, error: %d",errno));
427 result->pending=0; /* Assume everything is ok */
428 break;
429 }
430 ((my_aio_result*) tmp)->pending=0;
431 if ((my_aio_result*) tmp == result)
432 break;
433 }
434 }
435 return;
436 }
437 #endif
438
439
440 /*
441 Use this to reset cache to re-start reading or to change the type
442 between READ_CACHE <-> WRITE_CACHE
443 If we are doing a reinit of a cache where we have the start of the file
444 in the cache, we are reusing this memory without flushing it to disk.
445 */
446
reinit_io_cache(IO_CACHE * info,enum cache_type type,my_off_t seek_offset,my_bool use_async_io,my_bool clear_cache)447 my_bool reinit_io_cache(IO_CACHE *info, enum cache_type type,
448 my_off_t seek_offset,
449 my_bool use_async_io __attribute__((unused)),
450 my_bool clear_cache)
451 {
452 DBUG_ENTER("reinit_io_cache");
453 DBUG_PRINT("enter",("cache:%p type: %d seek_offset: %llu clear_cache: %d",
454 info, type, (ulonglong) seek_offset,
455 (int) clear_cache));
456
457 DBUG_ASSERT(type == READ_CACHE || type == WRITE_CACHE);
458 DBUG_ASSERT(info->type == READ_CACHE || info->type == WRITE_CACHE);
459
460 /* If the whole file is in memory, avoid flushing to disk */
461 if (! clear_cache &&
462 seek_offset >= info->pos_in_file &&
463 seek_offset <= my_b_tell(info))
464 {
465 /* Reuse current buffer without flushing it to disk */
466 uchar *pos;
467 if (info->type == WRITE_CACHE && type == READ_CACHE)
468 {
469 info->read_end=info->write_pos;
470 info->end_of_file=my_b_tell(info);
471 /*
472 Trigger a new seek only if we have a valid
473 file handle.
474 */
475 info->seek_not_done= (info->file != -1);
476 }
477 else if (type == WRITE_CACHE)
478 {
479 if (info->type == READ_CACHE)
480 {
481 info->write_end=info->write_buffer+info->buffer_length;
482 info->seek_not_done=1;
483 }
484 info->end_of_file = ~(my_off_t) 0;
485 }
486 pos=info->request_pos+(seek_offset-info->pos_in_file);
487 if (type == WRITE_CACHE)
488 info->write_pos=pos;
489 else
490 info->read_pos= pos;
491 #ifdef HAVE_AIOWAIT
492 my_aiowait(&info->aio_result); /* Wait for outstanding req */
493 #endif
494 }
495 else
496 {
497 /*
498 If we change from WRITE_CACHE to READ_CACHE, assume that everything
499 after the current positions should be ignored
500 */
501 if (info->type == WRITE_CACHE && type == READ_CACHE)
502 info->end_of_file=my_b_tell(info);
503 /* flush cache if we want to reuse it */
504 if (!clear_cache && my_b_flush_io_cache(info,1))
505 DBUG_RETURN(1);
506 info->pos_in_file=seek_offset;
507 /* Better to do always do a seek */
508 info->seek_not_done=1;
509 info->request_pos=info->read_pos=info->write_pos=info->buffer;
510 if (type == READ_CACHE)
511 {
512 info->read_end=info->buffer; /* Nothing in cache */
513 }
514 else
515 {
516 if (info->myflags & MY_ENCRYPT)
517 {
518 info->write_end = info->write_buffer + info->buffer_length;
519 if (seek_offset && info->file != -1)
520 {
521 info->read_end= info->buffer;
522 _my_b_encr_read(info, 0, 0); /* prefill the buffer */
523 info->write_pos= info->read_pos;
524 info->seek_not_done=1;
525 }
526 }
527 else
528 {
529 info->write_end=(info->buffer + info->buffer_length -
530 (seek_offset & (IO_SIZE-1)));
531 }
532 info->end_of_file= ~(my_off_t) 0;
533 }
534 }
535 info->type=type;
536 info->error=0;
537 init_functions(info);
538
539 #ifdef HAVE_AIOWAIT
540 if (use_async_io && ! my_disable_async_io &&
541 ((ulong) info->buffer_length <
542 (ulong) (info->end_of_file - seek_offset)))
543 {
544 DBUG_ASSERT(!(cache_myflags & MY_ENCRYPT));
545 info->read_length=info->buffer_length/2;
546 info->read_function=_my_b_async_read;
547 }
548 info->inited=0;
549 #endif
550 DBUG_RETURN(0);
551 } /* reinit_io_cache */
552
553
_my_b_read(IO_CACHE * info,uchar * Buffer,size_t Count)554 int _my_b_read(IO_CACHE *info, uchar *Buffer, size_t Count)
555 {
556 size_t left_length;
557 int res;
558
559 /* If the buffer is not empty yet, copy what is available. */
560 if ((left_length= (size_t) (info->read_end - info->read_pos)))
561 {
562 DBUG_ASSERT(Count > left_length);
563 memcpy(Buffer, info->read_pos, left_length);
564 Buffer+=left_length;
565 Count-=left_length;
566 }
567 res= info->read_function(info, Buffer, Count);
568 if (res && info->error >= 0)
569 info->error+= (int)left_length; /* update number or read bytes */
570 return res;
571 }
572
_my_b_write(IO_CACHE * info,const uchar * Buffer,size_t Count)573 int _my_b_write(IO_CACHE *info, const uchar *Buffer, size_t Count)
574 {
575 size_t rest_length;
576 int res;
577
578 /* Always use my_b_flush_io_cache() to flush write_buffer! */
579 DBUG_ASSERT(Buffer != info->write_buffer);
580
581 if (info->pos_in_file + info->buffer_length > info->end_of_file)
582 {
583 my_errno=errno=EFBIG;
584 return info->error = -1;
585 }
586
587 rest_length= (size_t) (info->write_end - info->write_pos);
588 DBUG_ASSERT(Count >= rest_length);
589 memcpy(info->write_pos, Buffer, (size_t) rest_length);
590 Buffer+=rest_length;
591 Count-=rest_length;
592 info->write_pos+=rest_length;
593
594 if (my_b_flush_io_cache(info, 1))
595 return 1;
596
597 if (Count)
598 {
599 my_off_t old_pos_in_file= info->pos_in_file;
600 res= info->write_function(info, Buffer, Count);
601 Count-= (size_t) (info->pos_in_file - old_pos_in_file);
602 Buffer+= info->pos_in_file - old_pos_in_file;
603 }
604 else
605 res= 0;
606
607 if (!res && Count)
608 {
609 memcpy(info->write_pos, Buffer, Count);
610 info->write_pos+= Count;
611 }
612 return res;
613 }
614
615 /*
616 Read buffered.
617
618 SYNOPSIS
619 _my_b_cache_read()
620 info IO_CACHE pointer
621 Buffer Buffer to retrieve count bytes from file
622 Count Number of bytes to read into Buffer
623
624 NOTE
625 This function is only called from the my_b_read() macro when there
626 isn't enough characters in the buffer to satisfy the request.
627
628 WARNING
629
630 When changing this function, be careful with handling file offsets
631 (end-of_file, pos_in_file). Do not cast them to possibly smaller
632 types than my_off_t unless you can be sure that their value fits.
633 Same applies to differences of file offsets.
634
635 When changing this function, check _my_b_cache_read_r(). It might need the
636 same change.
637
638 RETURN
639 0 we succeeded in reading all data
640 1 Error: couldn't read requested characters. In this case:
641 If info->error == -1, we got a read error.
642 Otherwise info->error contains the number of bytes in Buffer.
643 */
644
_my_b_cache_read(IO_CACHE * info,uchar * Buffer,size_t Count)645 int _my_b_cache_read(IO_CACHE *info, uchar *Buffer, size_t Count)
646 {
647 size_t length= 0, diff_length, left_length= 0, max_length;
648 my_off_t pos_in_file;
649 DBUG_ENTER("_my_b_cache_read");
650
651 /* pos_in_file always point on where info->buffer was read */
652 pos_in_file=info->pos_in_file+ (size_t) (info->read_end - info->buffer);
653
654 /*
655 Whenever a function which operates on IO_CACHE flushes/writes
656 some part of the IO_CACHE to disk it will set the property
657 "seek_not_done" to indicate this to other functions operating
658 on the IO_CACHE.
659 */
660 if (info->seek_not_done)
661 {
662 if ((mysql_file_seek(info->file, pos_in_file, MY_SEEK_SET, MYF(0))
663 != MY_FILEPOS_ERROR))
664 {
665 /* No error, reset seek_not_done flag. */
666 info->seek_not_done= 0;
667
668 if (info->next_file_user)
669 {
670 IO_CACHE *c;
671 for (c= info->next_file_user;
672 c!= info;
673 c= c->next_file_user)
674 {
675 c->seek_not_done= 1;
676 }
677 }
678 }
679 else
680 {
681 /*
682 If the seek failed and the error number is ESPIPE, it is because
683 info->file is a pipe or socket or FIFO. We never should have tried
684 to seek on that. See Bugs#25807 and #22828 for more info.
685 */
686 DBUG_ASSERT(my_errno != ESPIPE);
687 info->error= -1;
688 DBUG_RETURN(1);
689 }
690 }
691
692 /*
693 Calculate, how much we are within a IO_SIZE block. Ideally this
694 should be zero.
695 */
696 diff_length= (size_t) (pos_in_file & (IO_SIZE-1));
697
698 /*
699 If more than a block plus the rest of the current block is wanted,
700 we do read directly, without filling the buffer.
701 */
702 if (Count >= (size_t) (IO_SIZE+(IO_SIZE-diff_length)))
703 { /* Fill first intern buffer */
704 size_t read_length;
705 if (info->end_of_file <= pos_in_file)
706 {
707 /* End of file. Return, what we did copy from the buffer. */
708 info->error= (int) left_length;
709 info->seek_not_done=1;
710 DBUG_RETURN(1);
711 }
712 /*
713 Crop the wanted count to a multiple of IO_SIZE and subtract,
714 what we did already read from a block. That way, the read will
715 end aligned with a block.
716 */
717 length= IO_ROUND_DN(Count) - diff_length;
718 if ((read_length= mysql_file_read(info->file,Buffer, length, info->myflags))
719 != length)
720 {
721 /*
722 If we didn't get, what we wanted, we either return -1 for a read
723 error, or (it's end of file), how much we got in total.
724 */
725 info->error= (read_length == (size_t) -1 ? -1 :
726 (int) (read_length+left_length));
727 info->seek_not_done=1;
728 DBUG_RETURN(1);
729 }
730 Count-=length;
731 Buffer+=length;
732 pos_in_file+=length;
733 left_length+=length;
734 diff_length=0;
735 }
736
737 /*
738 At this point, we want less than one and a partial block.
739 We will read a full cache, minus the number of bytes, we are
740 within a block already. So we will reach new alignment.
741 */
742 max_length= info->read_length-diff_length;
743 /* We will not read past end of file. */
744 if (info->type != READ_FIFO &&
745 max_length > (info->end_of_file - pos_in_file))
746 max_length= (size_t) (info->end_of_file - pos_in_file);
747 /*
748 If there is nothing left to read,
749 we either are done, or we failed to fulfill the request.
750 Otherwise, we read max_length into the cache.
751 */
752 if (!max_length)
753 {
754 if (Count)
755 {
756 /* We couldn't fulfil the request. Return, how much we got. */
757 info->error= (int) left_length;
758 DBUG_RETURN(1);
759 }
760 else
761 {
762 info->error= 0;
763 if (length == 0) /* nothing was read */
764 DBUG_RETURN(0); /* EOF */
765
766 length= 0; /* non-zero size read was done */
767 }
768 }
769 else
770 {
771 if (info->next_file_user)
772 {
773 IO_CACHE *c;
774 for (c= info->next_file_user;
775 c!= info;
776 c= c->next_file_user)
777 {
778 c->seek_not_done= 1;
779 }
780 }
781 if ((length= mysql_file_read(info->file,info->buffer, max_length,
782 info->myflags)) < Count ||
783 length == (size_t) -1)
784 {
785 /*
786 We got an read error, or less than requested (end of file).
787 If not a read error, copy, what we got.
788 */
789 if (length != (size_t) -1)
790 memcpy(Buffer, info->buffer, length);
791 info->pos_in_file= pos_in_file;
792 /* For a read error, return -1, otherwise, what we got in total. */
793 info->error= length == (size_t) -1 ? -1 : (int) (length+left_length);
794 info->read_pos=info->read_end=info->buffer;
795 info->seek_not_done=1;
796 DBUG_RETURN(1);
797 }
798 }
799 /*
800 Count is the remaining number of bytes requested.
801 length is the amount of data in the cache.
802 Read Count bytes from the cache.
803 */
804 info->read_pos=info->buffer+Count;
805 info->read_end=info->buffer+length;
806 info->pos_in_file=pos_in_file;
807 if (Count)
808 memcpy(Buffer, info->buffer, Count);
809 DBUG_RETURN(0);
810 }
811
812
813 /*
814 Prepare IO_CACHE for shared use.
815
816 SYNOPSIS
817 init_io_cache_share()
818 read_cache A read cache. This will be copied for
819 every thread after setup.
820 cshare The share.
821 write_cache If non-NULL a write cache that is to be
822 synchronized with the read caches.
823 num_threads Number of threads sharing the cache
824 including the write thread if any.
825
826 DESCRIPTION
827
828 The shared cache is used so: One IO_CACHE is initialized with
829 init_io_cache(). This includes the allocation of a buffer. Then a
830 share is allocated and init_io_cache_share() is called with the io
831 cache and the share. Then the io cache is copied for each thread. So
832 every thread has its own copy of IO_CACHE. But the allocated buffer
833 is shared because cache->buffer is the same for all caches.
834
835 One thread reads data from the file into the buffer. All threads
836 read from the buffer, but every thread maintains its own set of
837 pointers into the buffer. When all threads have used up the buffer
838 contents, one of the threads reads the next block of data into the
839 buffer. To accomplish this, each thread enters the cache lock before
840 accessing the buffer. They wait in lock_io_cache() until all threads
841 joined the lock. The last thread entering the lock is in charge of
842 reading from file to buffer. It wakes all threads when done.
843
844 Synchronizing a write cache to the read caches works so: Whenever
845 the write buffer needs a flush, the write thread enters the lock and
846 waits for all other threads to enter the lock too. They do this when
847 they have used up the read buffer. When all threads are in the lock,
848 the write thread copies the write buffer to the read buffer and
849 wakes all threads.
850
851 share->running_threads is the number of threads not being in the
852 cache lock. When entering lock_io_cache() the number is decreased.
853 When the thread that fills the buffer enters unlock_io_cache() the
854 number is reset to the number of threads. The condition
855 running_threads == 0 means that all threads are in the lock. Bumping
856 up the number to the full count is non-intuitive. But increasing the
857 number by one for each thread that leaves the lock could lead to a
858 solo run of one thread. The last thread to join a lock reads from
859 file to buffer, wakes the other threads, processes the data in the
860 cache and enters the lock again. If no other thread left the lock
861 meanwhile, it would think it's the last one again and read the next
862 block...
863
864 The share has copies of 'error', 'buffer', 'read_end', and
865 'pos_in_file' from the thread that filled the buffer. We may not be
866 able to access this information directly from its cache because the
867 thread may be removed from the share before the variables could be
868 copied by all other threads. Or, if a write buffer is synchronized,
869 it would change its 'pos_in_file' after waking the other threads,
870 possibly before they could copy its value.
871
872 However, the 'buffer' variable in the share is for a synchronized
873 write cache. It needs to know where to put the data. Otherwise it
874 would need access to the read cache of one of the threads that is
875 not yet removed from the share.
876
877 RETURN
878 void
879 */
880
init_io_cache_share(IO_CACHE * read_cache,IO_CACHE_SHARE * cshare,IO_CACHE * write_cache,uint num_threads)881 void init_io_cache_share(IO_CACHE *read_cache, IO_CACHE_SHARE *cshare,
882 IO_CACHE *write_cache, uint num_threads)
883 {
884 DBUG_ENTER("init_io_cache_share");
885 DBUG_PRINT("io_cache_share", ("read_cache: %p share: %p "
886 "write_cache: %p threads: %u",
887 read_cache, cshare,
888 write_cache, num_threads));
889
890 DBUG_ASSERT(num_threads > 1);
891 DBUG_ASSERT(read_cache->type == READ_CACHE);
892 DBUG_ASSERT(!write_cache || (write_cache->type == WRITE_CACHE));
893
894 mysql_mutex_init(key_IO_CACHE_SHARE_mutex,
895 &cshare->mutex, MY_MUTEX_INIT_FAST);
896 mysql_cond_init(key_IO_CACHE_SHARE_cond, &cshare->cond, 0);
897 mysql_cond_init(key_IO_CACHE_SHARE_cond_writer, &cshare->cond_writer, 0);
898
899 cshare->running_threads= num_threads;
900 cshare->total_threads= num_threads;
901 cshare->error= 0; /* Initialize. */
902 cshare->buffer= read_cache->buffer;
903 cshare->read_end= NULL; /* See function comment of lock_io_cache(). */
904 cshare->pos_in_file= 0; /* See function comment of lock_io_cache(). */
905 cshare->source_cache= write_cache; /* Can be NULL. */
906
907 read_cache->share= cshare;
908 read_cache->read_function= _my_b_cache_read_r;
909
910 if (write_cache)
911 {
912 write_cache->share= cshare;
913 write_cache->write_function= _my_b_cache_write_r;
914 }
915
916 DBUG_VOID_RETURN;
917 }
918
919
920 /*
921 Remove a thread from shared access to IO_CACHE.
922
923 SYNOPSIS
924 remove_io_thread()
925 cache The IO_CACHE to be removed from the share.
926
927 NOTE
928
929 Every thread must do that on exit for not to deadlock other threads.
930
931 The last thread destroys the pthread resources.
932
933 A writer flushes its cache first.
934
935 RETURN
936 void
937 */
938
remove_io_thread(IO_CACHE * cache)939 void remove_io_thread(IO_CACHE *cache)
940 {
941 IO_CACHE_SHARE *cshare= cache->share;
942 uint total;
943 DBUG_ENTER("remove_io_thread");
944
945 /* If the writer goes, it needs to flush the write cache. */
946 if (cache == cshare->source_cache)
947 flush_io_cache(cache);
948
949 mysql_mutex_lock(&cshare->mutex);
950 DBUG_PRINT("io_cache_share", ("%s: %p",
951 (cache == cshare->source_cache) ?
952 "writer" : "reader", cache));
953
954 /* Remove from share. */
955 total= --cshare->total_threads;
956 DBUG_PRINT("io_cache_share", ("remaining threads: %u", total));
957
958 /* Detach from share. */
959 cache->share= NULL;
960
961 /* If the writer goes, let the readers know. */
962 if (cache == cshare->source_cache)
963 {
964 DBUG_PRINT("io_cache_share", ("writer leaves"));
965 cshare->source_cache= NULL;
966 }
967
968 /* If all threads are waiting for me to join the lock, wake them. */
969 if (!--cshare->running_threads)
970 {
971 DBUG_PRINT("io_cache_share", ("the last running thread leaves, wake all"));
972 mysql_cond_signal(&cshare->cond_writer);
973 mysql_cond_broadcast(&cshare->cond);
974 }
975
976 mysql_mutex_unlock(&cshare->mutex);
977
978 if (!total)
979 {
980 DBUG_PRINT("io_cache_share", ("last thread removed, destroy share"));
981 mysql_cond_destroy (&cshare->cond_writer);
982 mysql_cond_destroy (&cshare->cond);
983 mysql_mutex_destroy(&cshare->mutex);
984 }
985
986 DBUG_VOID_RETURN;
987 }
988
989
990 /*
991 Lock IO cache and wait for all other threads to join.
992
993 SYNOPSIS
994 lock_io_cache()
995 cache The cache of the thread entering the lock.
996 pos File position of the block to read.
997 Unused for the write thread.
998
999 DESCRIPTION
1000
1001 Wait for all threads to finish with the current buffer. We want
1002 all threads to proceed in concert. The last thread to join
1003 lock_io_cache() will read the block from file and all threads start
1004 to use it. Then they will join again for reading the next block.
1005
1006 The waiting threads detect a fresh buffer by comparing
1007 cshare->pos_in_file with the position they want to process next.
1008 Since the first block may start at position 0, we take
1009 cshare->read_end as an additional condition. This variable is
1010 initialized to NULL and will be set after a block of data is written
1011 to the buffer.
1012
1013 RETURN
1014 1 OK, lock in place, go ahead and read.
1015 0 OK, unlocked, another thread did the read.
1016 */
1017
lock_io_cache(IO_CACHE * cache,my_off_t pos)1018 static int lock_io_cache(IO_CACHE *cache, my_off_t pos)
1019 {
1020 IO_CACHE_SHARE *cshare= cache->share;
1021 DBUG_ENTER("lock_io_cache");
1022
1023 /* Enter the lock. */
1024 mysql_mutex_lock(&cshare->mutex);
1025 cshare->running_threads--;
1026 DBUG_PRINT("io_cache_share", ("%s: %p pos: %lu running: %u",
1027 (cache == cshare->source_cache) ?
1028 "writer" : "reader", cache, (ulong) pos,
1029 cshare->running_threads));
1030
1031 if (cshare->source_cache)
1032 {
1033 /* A write cache is synchronized to the read caches. */
1034
1035 if (cache == cshare->source_cache)
1036 {
1037 /* The writer waits until all readers are here. */
1038 while (cshare->running_threads)
1039 {
1040 DBUG_PRINT("io_cache_share", ("writer waits in lock"));
1041 mysql_cond_wait(&cshare->cond_writer, &cshare->mutex);
1042 }
1043 DBUG_PRINT("io_cache_share", ("writer awoke, going to copy"));
1044
1045 /* Stay locked. Leave the lock later by unlock_io_cache(). */
1046 DBUG_RETURN(1);
1047 }
1048
1049 /* The last thread wakes the writer. */
1050 if (!cshare->running_threads)
1051 {
1052 DBUG_PRINT("io_cache_share", ("waking writer"));
1053 mysql_cond_signal(&cshare->cond_writer);
1054 }
1055
1056 /*
1057 Readers wait until the data is copied from the writer. Another
1058 reason to stop waiting is the removal of the write thread. If this
1059 happens, we leave the lock with old data in the buffer.
1060 */
1061 while ((!cshare->read_end || (cshare->pos_in_file < pos)) &&
1062 cshare->source_cache)
1063 {
1064 DBUG_PRINT("io_cache_share", ("reader waits in lock"));
1065 mysql_cond_wait(&cshare->cond, &cshare->mutex);
1066 }
1067
1068 /*
1069 If the writer was removed from the share while this thread was
1070 asleep, we need to simulate an EOF condition. The writer cannot
1071 reset the share variables as they might still be in use by readers
1072 of the last block. When we awake here then because the last
1073 joining thread signalled us. If the writer is not the last, it
1074 will not signal. So it is safe to clear the buffer here.
1075 */
1076 if (!cshare->read_end || (cshare->pos_in_file < pos))
1077 {
1078 DBUG_PRINT("io_cache_share", ("reader found writer removed. EOF"));
1079 cshare->read_end= cshare->buffer; /* Empty buffer. */
1080 cshare->error= 0; /* EOF is not an error. */
1081 }
1082 }
1083 else
1084 {
1085 /*
1086 There are read caches only. The last thread arriving in
1087 lock_io_cache() continues with a locked cache and reads the block.
1088 */
1089 if (!cshare->running_threads)
1090 {
1091 DBUG_PRINT("io_cache_share", ("last thread joined, going to read"));
1092 /* Stay locked. Leave the lock later by unlock_io_cache(). */
1093 DBUG_RETURN(1);
1094 }
1095
1096 /*
1097 All other threads wait until the requested block is read by the
1098 last thread arriving. Another reason to stop waiting is the
1099 removal of a thread. If this leads to all threads being in the
1100 lock, we have to continue also. The first of the awaken threads
1101 will then do the read.
1102 */
1103 while ((!cshare->read_end || (cshare->pos_in_file < pos)) &&
1104 cshare->running_threads)
1105 {
1106 DBUG_PRINT("io_cache_share", ("reader waits in lock"));
1107 mysql_cond_wait(&cshare->cond, &cshare->mutex);
1108 }
1109
1110 /* If the block is not yet read, continue with a locked cache and read. */
1111 if (!cshare->read_end || (cshare->pos_in_file < pos))
1112 {
1113 DBUG_PRINT("io_cache_share", ("reader awoke, going to read"));
1114 /* Stay locked. Leave the lock later by unlock_io_cache(). */
1115 DBUG_RETURN(1);
1116 }
1117
1118 /* Another thread did read the block already. */
1119 }
1120 DBUG_PRINT("io_cache_share", ("reader awoke, going to process %u bytes",
1121 (uint) (cshare->read_end ? (size_t)
1122 (cshare->read_end - cshare->buffer) :
1123 0)));
1124
1125 /*
1126 Leave the lock. Do not call unlock_io_cache() later. The thread that
1127 filled the buffer did this and marked all threads as running.
1128 */
1129 mysql_mutex_unlock(&cshare->mutex);
1130 DBUG_RETURN(0);
1131 }
1132
1133
1134 /*
1135 Unlock IO cache.
1136
1137 SYNOPSIS
1138 unlock_io_cache()
1139 cache The cache of the thread leaving the lock.
1140
1141 NOTE
1142 This is called by the thread that filled the buffer. It marks all
1143 threads as running and awakes them. This must not be done by any
1144 other thread.
1145
1146 Do not signal cond_writer. Either there is no writer or the writer
1147 is the only one who can call this function.
1148
1149 The reason for resetting running_threads to total_threads before
1150 waking all other threads is that it could be possible that this
1151 thread is so fast with processing the buffer that it enters the lock
1152 before even one other thread has left it. If every awoken thread
1153 would increase running_threads by one, this thread could think that
1154 he is again the last to join and would not wait for the other
1155 threads to process the data.
1156
1157 RETURN
1158 void
1159 */
1160
unlock_io_cache(IO_CACHE * cache)1161 static void unlock_io_cache(IO_CACHE *cache)
1162 {
1163 IO_CACHE_SHARE *cshare= cache->share;
1164 DBUG_ENTER("unlock_io_cache");
1165 DBUG_PRINT("io_cache_share", ("%s: %p pos: %lu running: %u",
1166 (cache == cshare->source_cache) ?
1167 "writer" : "reader",
1168 cache, (ulong) cshare->pos_in_file,
1169 cshare->total_threads));
1170
1171 cshare->running_threads= cshare->total_threads;
1172 mysql_cond_broadcast(&cshare->cond);
1173 mysql_mutex_unlock(&cshare->mutex);
1174 DBUG_VOID_RETURN;
1175 }
1176
1177
1178 /*
1179 Read from IO_CACHE when it is shared between several threads.
1180
1181 SYNOPSIS
1182 _my_b_cache_read_r()
1183 cache IO_CACHE pointer
1184 Buffer Buffer to retrieve count bytes from file
1185 Count Number of bytes to read into Buffer
1186
1187 NOTE
1188 This function is only called from the my_b_read() macro when there
1189 isn't enough characters in the buffer to satisfy the request.
1190
1191 IMPLEMENTATION
1192
1193 It works as follows: when a thread tries to read from a file (that
1194 is, after using all the data from the (shared) buffer), it just
1195 hangs on lock_io_cache(), waiting for other threads. When the very
1196 last thread attempts a read, lock_io_cache() returns 1, the thread
1197 does actual IO and unlock_io_cache(), which signals all the waiting
1198 threads that data is in the buffer.
1199
1200 WARNING
1201
1202 When changing this function, be careful with handling file offsets
1203 (end-of_file, pos_in_file). Do not cast them to possibly smaller
1204 types than my_off_t unless you can be sure that their value fits.
1205 Same applies to differences of file offsets. (Bug #11527)
1206
1207 When changing this function, check _my_b_cache_read(). It might need the
1208 same change.
1209
1210 RETURN
1211 0 we succeeded in reading all data
1212 1 Error: can't read requested characters
1213 */
1214
_my_b_cache_read_r(IO_CACHE * cache,uchar * Buffer,size_t Count)1215 static int _my_b_cache_read_r(IO_CACHE *cache, uchar *Buffer, size_t Count)
1216 {
1217 my_off_t pos_in_file;
1218 size_t length, diff_length, left_length= 0;
1219 IO_CACHE_SHARE *cshare= cache->share;
1220 DBUG_ENTER("_my_b_cache_read_r");
1221 DBUG_ASSERT(!(cache->myflags & MY_ENCRYPT));
1222
1223 while (Count)
1224 {
1225 size_t cnt, len;
1226
1227 pos_in_file= cache->pos_in_file + (cache->read_end - cache->buffer);
1228 diff_length= (size_t) (pos_in_file & (IO_SIZE-1));
1229 length=IO_ROUND_UP(Count+diff_length)-diff_length;
1230 length= ((length <= cache->read_length) ?
1231 length + IO_ROUND_DN(cache->read_length - length) :
1232 length - IO_ROUND_UP(length - cache->read_length));
1233 if (cache->type != READ_FIFO &&
1234 (length > (cache->end_of_file - pos_in_file)))
1235 length= (size_t) (cache->end_of_file - pos_in_file);
1236 if (length == 0)
1237 {
1238 cache->error= (int) left_length;
1239 DBUG_RETURN(1);
1240 }
1241 if (lock_io_cache(cache, pos_in_file))
1242 {
1243 /* With a synchronized write/read cache we won't come here... */
1244 DBUG_ASSERT(!cshare->source_cache);
1245 /*
1246 ... unless the writer has gone before this thread entered the
1247 lock. Simulate EOF in this case. It can be distinguished by
1248 cache->file.
1249 */
1250 if (cache->file < 0)
1251 len= 0;
1252 else
1253 {
1254 /*
1255 Whenever a function which operates on IO_CACHE flushes/writes
1256 some part of the IO_CACHE to disk it will set the property
1257 "seek_not_done" to indicate this to other functions operating
1258 on the IO_CACHE.
1259 */
1260 if (cache->seek_not_done)
1261 {
1262 if (mysql_file_seek(cache->file, pos_in_file, MY_SEEK_SET, MYF(0))
1263 == MY_FILEPOS_ERROR)
1264 {
1265 cache->error= -1;
1266 unlock_io_cache(cache);
1267 DBUG_RETURN(1);
1268 }
1269 }
1270 len= mysql_file_read(cache->file, cache->buffer, length, cache->myflags);
1271 }
1272 DBUG_PRINT("io_cache_share", ("read %lu bytes", (ulong) len));
1273
1274 cache->read_end= cache->buffer + (len == (size_t) -1 ? 0 : len);
1275 cache->error= (len == length ? 0 : (int) len);
1276 cache->pos_in_file= pos_in_file;
1277
1278 /* Copy important values to the share. */
1279 cshare->error= cache->error;
1280 cshare->read_end= cache->read_end;
1281 cshare->pos_in_file= pos_in_file;
1282
1283 /* Mark all threads as running and wake them. */
1284 unlock_io_cache(cache);
1285 }
1286 else
1287 {
1288 /*
1289 With a synchronized write/read cache readers always come here.
1290 Copy important values from the share.
1291 */
1292 cache->error= cshare->error;
1293 cache->read_end= cshare->read_end;
1294 cache->pos_in_file= cshare->pos_in_file;
1295
1296 len= ((cache->error == -1) ? (size_t) -1 :
1297 (size_t) (cache->read_end - cache->buffer));
1298 }
1299 cache->read_pos= cache->buffer;
1300 cache->seek_not_done= 0;
1301 if (len == 0 || len == (size_t) -1)
1302 {
1303 DBUG_PRINT("io_cache_share", ("reader error. len %lu left %lu",
1304 (ulong) len, (ulong) left_length));
1305 cache->error= (int) left_length;
1306 DBUG_RETURN(1);
1307 }
1308 cnt= (len > Count) ? Count : len;
1309 if (cnt)
1310 memcpy(Buffer, cache->read_pos, cnt);
1311 Count -= cnt;
1312 Buffer+= cnt;
1313 left_length+= cnt;
1314 cache->read_pos+= cnt;
1315 }
1316 DBUG_RETURN(0);
1317 }
1318
1319
1320 /*
1321 Copy data from write cache to read cache.
1322
1323 SYNOPSIS
1324 copy_to_read_buffer()
1325 write_cache The write cache.
1326 write_buffer The source of data, mostly the cache buffer.
1327 write_length The number of bytes to copy.
1328
1329 NOTE
1330 The write thread will wait for all read threads to join the cache
1331 lock. Then it copies the data over and wakes the read threads.
1332
1333 RETURN
1334 void
1335 */
1336
copy_to_read_buffer(IO_CACHE * write_cache,const uchar * write_buffer,my_off_t pos_in_file)1337 static void copy_to_read_buffer(IO_CACHE *write_cache,
1338 const uchar *write_buffer, my_off_t pos_in_file)
1339 {
1340 size_t write_length= (size_t) (write_cache->pos_in_file - pos_in_file);
1341 IO_CACHE_SHARE *cshare= write_cache->share;
1342
1343 DBUG_ASSERT(cshare->source_cache == write_cache);
1344 /*
1345 write_length is usually less or equal to buffer_length.
1346 It can be bigger if _my_b_cache_write_r() is called with a big length.
1347 */
1348 while (write_length)
1349 {
1350 size_t copy_length= MY_MIN(write_length, write_cache->buffer_length);
1351 int __attribute__((unused)) rc;
1352
1353 rc= lock_io_cache(write_cache, pos_in_file);
1354 /* The writing thread does always have the lock when it awakes. */
1355 DBUG_ASSERT(rc);
1356
1357 memcpy(cshare->buffer, write_buffer, copy_length);
1358
1359 cshare->error= 0;
1360 cshare->read_end= cshare->buffer + copy_length;
1361 cshare->pos_in_file= pos_in_file;
1362
1363 /* Mark all threads as running and wake them. */
1364 unlock_io_cache(write_cache);
1365
1366 write_buffer+= copy_length;
1367 write_length-= copy_length;
1368 }
1369 }
1370
1371
1372 /*
1373 Do sequential read from the SEQ_READ_APPEND cache.
1374
1375 We do this in three stages:
1376 - first read from info->buffer
1377 - then if there are still data to read, try the file descriptor
1378 - afterwards, if there are still data to read, try append buffer
1379
1380 RETURNS
1381 0 Success
1382 1 Failed to read
1383 */
1384
_my_b_seq_read(IO_CACHE * info,uchar * Buffer,size_t Count)1385 static int _my_b_seq_read(IO_CACHE *info, uchar *Buffer, size_t Count)
1386 {
1387 size_t length, diff_length, save_count, max_length;
1388 my_off_t pos_in_file;
1389 save_count=Count;
1390
1391 lock_append_buffer(info);
1392
1393 /* pos_in_file always point on where info->buffer was read */
1394 if ((pos_in_file=info->pos_in_file +
1395 (size_t) (info->read_end - info->buffer)) >= info->end_of_file)
1396 goto read_append_buffer;
1397
1398 /*
1399 With read-append cache we must always do a seek before we read,
1400 because the write could have moved the file pointer astray
1401 */
1402 if (mysql_file_seek(info->file, pos_in_file, MY_SEEK_SET, MYF(0)) == MY_FILEPOS_ERROR)
1403 {
1404 info->error= -1;
1405 unlock_append_buffer(info);
1406 return (1);
1407 }
1408 info->seek_not_done=0;
1409
1410 diff_length= (size_t) (pos_in_file & (IO_SIZE-1));
1411
1412 /* now the second stage begins - read from file descriptor */
1413 if (Count >= (size_t) (IO_SIZE+(IO_SIZE-diff_length)))
1414 {
1415 /* Fill first intern buffer */
1416 size_t read_length;
1417
1418 length= IO_ROUND_DN(Count) - diff_length;
1419 if ((read_length= mysql_file_read(info->file,Buffer, length,
1420 info->myflags)) == (size_t) -1)
1421 {
1422 info->error= -1;
1423 unlock_append_buffer(info);
1424 return 1;
1425 }
1426 Count-=read_length;
1427 Buffer+=read_length;
1428 pos_in_file+=read_length;
1429
1430 if (read_length != length)
1431 {
1432 /*
1433 We only got part of data; Read the rest of the data from the
1434 write buffer
1435 */
1436 goto read_append_buffer;
1437 }
1438 diff_length=0;
1439 }
1440
1441 max_length= info->read_length-diff_length;
1442 if (max_length > (info->end_of_file - pos_in_file))
1443 max_length= (size_t) (info->end_of_file - pos_in_file);
1444 if (!max_length)
1445 {
1446 if (Count)
1447 goto read_append_buffer;
1448 length=0; /* Didn't read any more chars */
1449 }
1450 else
1451 {
1452 length= mysql_file_read(info->file,info->buffer, max_length, info->myflags);
1453 if (length == (size_t) -1)
1454 {
1455 info->error= -1;
1456 unlock_append_buffer(info);
1457 return 1;
1458 }
1459 if (length < Count)
1460 {
1461 memcpy(Buffer, info->buffer, length);
1462 Count -= length;
1463 Buffer += length;
1464
1465 /*
1466 added the line below to make
1467 DBUG_ASSERT(pos_in_file==info->end_of_file) pass.
1468 otherwise this does not appear to be needed
1469 */
1470 pos_in_file += length;
1471 goto read_append_buffer;
1472 }
1473 }
1474 unlock_append_buffer(info);
1475 info->read_pos=info->buffer+Count;
1476 info->read_end=info->buffer+length;
1477 info->pos_in_file=pos_in_file;
1478 memcpy(Buffer,info->buffer,(size_t) Count);
1479 return 0;
1480
1481 read_append_buffer:
1482
1483 /*
1484 Read data from the current write buffer.
1485 Count should never be == 0 here (The code will work even if count is 0)
1486 */
1487
1488 {
1489 /* First copy the data to Count */
1490 size_t len_in_buff = (size_t) (info->write_pos - info->append_read_pos);
1491 size_t copy_len;
1492 size_t transfer_len;
1493
1494 DBUG_ASSERT(info->append_read_pos <= info->write_pos);
1495 copy_len=MY_MIN(Count, len_in_buff);
1496 memcpy(Buffer, info->append_read_pos, copy_len);
1497 info->append_read_pos += copy_len;
1498 Count -= copy_len;
1499 if (Count)
1500 info->error= (int) (save_count - Count);
1501
1502 /* Fill read buffer with data from write buffer */
1503 memcpy(info->buffer, info->append_read_pos,
1504 (size_t) (transfer_len=len_in_buff - copy_len));
1505 info->read_pos= info->buffer;
1506 info->read_end= info->buffer+transfer_len;
1507 info->append_read_pos=info->write_pos;
1508 info->pos_in_file=pos_in_file+copy_len;
1509 info->end_of_file+=len_in_buff;
1510 }
1511 unlock_append_buffer(info);
1512 return Count ? 1 : 0;
1513 }
1514
1515
1516 #ifdef HAVE_AIOWAIT
1517
1518 /*
1519 Read from the IO_CACHE into a buffer and feed asynchronously
1520 from disk when needed.
1521
1522 SYNOPSIS
1523 _my_b_async_read()
1524 info IO_CACHE pointer
1525 Buffer Buffer to retrieve count bytes from file
1526 Count Number of bytes to read into Buffer
1527
1528 RETURN VALUE
1529 -1 An error has occurred; my_errno is set.
1530 0 Success
1531 1 An error has occurred; IO_CACHE to error state.
1532 */
1533
_my_b_async_read(IO_CACHE * info,uchar * Buffer,size_t Count)1534 int _my_b_async_read(IO_CACHE *info, uchar *Buffer, size_t Count)
1535 {
1536 size_t length, read_length, diff_length, left_length=0, use_length, org_Count;
1537 size_t max_length;
1538 my_off_t next_pos_in_file;
1539 uchar *read_buffer;
1540
1541 org_Count=Count;
1542
1543 if (info->inited)
1544 { /* wait for read block */
1545 info->inited=0; /* No more block to read */
1546 my_aiowait(&info->aio_result); /* Wait for outstanding req */
1547 if (info->aio_result.result.aio_errno)
1548 {
1549 if (info->myflags & MY_WME)
1550 my_error(EE_READ, MYF(ME_BELL+ME_WAITTANG),
1551 my_filename(info->file),
1552 info->aio_result.result.aio_errno);
1553 my_errno=info->aio_result.result.aio_errno;
1554 info->error= -1;
1555 return(1);
1556 }
1557 if (! (read_length= (size_t) info->aio_result.result.aio_return) ||
1558 read_length == (size_t) -1)
1559 {
1560 my_errno=0; /* For testing */
1561 info->error= (read_length == (size_t) -1 ? -1 :
1562 (int) (read_length+left_length));
1563 return(1);
1564 }
1565 info->pos_in_file+= (size_t) (info->read_end - info->request_pos);
1566
1567 if (info->request_pos != info->buffer)
1568 info->request_pos=info->buffer;
1569 else
1570 info->request_pos=info->buffer+info->read_length;
1571 info->read_pos=info->request_pos;
1572 next_pos_in_file=info->aio_read_pos+read_length;
1573
1574 /* Check if pos_in_file is changed
1575 (_ni_read_cache may have skipped some bytes) */
1576
1577 if (info->aio_read_pos < info->pos_in_file)
1578 { /* Fix if skipped bytes */
1579 if (info->aio_read_pos + read_length < info->pos_in_file)
1580 {
1581 read_length=0; /* Skip block */
1582 next_pos_in_file=info->pos_in_file;
1583 }
1584 else
1585 {
1586 my_off_t offset= (info->pos_in_file - info->aio_read_pos);
1587 info->pos_in_file=info->aio_read_pos; /* Whe are here */
1588 info->read_pos=info->request_pos+offset;
1589 read_length-=offset; /* Bytes left from read_pos */
1590 }
1591 }
1592 #ifndef DBUG_OFF
1593 if (info->aio_read_pos > info->pos_in_file)
1594 {
1595 my_errno=EINVAL;
1596 return(info->read_length= (size_t) -1);
1597 }
1598 #endif
1599 /* Copy found bytes to buffer */
1600 length=MY_MIN(Count,read_length);
1601 memcpy(Buffer,info->read_pos,(size_t) length);
1602 Buffer+=length;
1603 Count-=length;
1604 left_length+=length;
1605 info->read_end=info->read_pos+read_length;
1606 info->read_pos+=length;
1607 }
1608 else
1609 next_pos_in_file=(info->pos_in_file+ (size_t)
1610 (info->read_end - info->request_pos));
1611
1612 /* If reading large blocks, or first read or read with skip */
1613 if (Count)
1614 {
1615 if (next_pos_in_file == info->end_of_file)
1616 {
1617 info->error=(int) (read_length+left_length);
1618 return 1;
1619 }
1620
1621 if (mysql_file_seek(info->file, next_pos_in_file, MY_SEEK_SET, MYF(0))
1622 == MY_FILEPOS_ERROR)
1623 {
1624 info->error= -1;
1625 return (1);
1626 }
1627
1628 read_length=IO_SIZE*2- (size_t) (next_pos_in_file & (IO_SIZE-1));
1629 if (Count < read_length)
1630 { /* Small block, read to cache */
1631 if ((read_length=mysql_file_read(info->file,info->request_pos,
1632 read_length, info->myflags)) == (size_t) -1)
1633 return info->error= -1;
1634 use_length=MY_MIN(Count,read_length);
1635 memcpy(Buffer,info->request_pos,(size_t) use_length);
1636 info->read_pos=info->request_pos+Count;
1637 info->read_end=info->request_pos+read_length;
1638 info->pos_in_file=next_pos_in_file; /* Start of block in cache */
1639 next_pos_in_file+=read_length;
1640
1641 if (Count != use_length)
1642 { /* Didn't find hole block */
1643 if (info->myflags & (MY_WME | MY_FAE | MY_FNABP) && Count != org_Count)
1644 my_error(EE_EOFERR, MYF(ME_BELL+ME_WAITTANG),
1645 my_filename(info->file),my_errno);
1646 info->error=(int) (read_length+left_length);
1647 return 1;
1648 }
1649 }
1650 else
1651 { /* Big block, don't cache it */
1652 if ((read_length= mysql_file_read(info->file, Buffer, Count,info->myflags))
1653 != Count)
1654 {
1655 info->error= read_length == (size_t) -1 ? -1 : read_length+left_length;
1656 return 1;
1657 }
1658 info->read_pos=info->read_end=info->request_pos;
1659 info->pos_in_file=(next_pos_in_file+=Count);
1660 }
1661 }
1662
1663 /* Read next block with asyncronic io */
1664 diff_length=(next_pos_in_file & (IO_SIZE-1));
1665 max_length= info->read_length - diff_length;
1666 if (max_length > info->end_of_file - next_pos_in_file)
1667 max_length= (size_t) (info->end_of_file - next_pos_in_file);
1668
1669 if (info->request_pos != info->buffer)
1670 read_buffer=info->buffer;
1671 else
1672 read_buffer=info->buffer+info->read_length;
1673 info->aio_read_pos=next_pos_in_file;
1674 if (max_length)
1675 {
1676 info->aio_result.result.aio_errno=AIO_INPROGRESS; /* Marker for test */
1677 DBUG_PRINT("aioread",("filepos: %ld length: %lu",
1678 (ulong) next_pos_in_file, (ulong) max_length));
1679 if (aioread(info->file,read_buffer, max_length,
1680 (my_off_t) next_pos_in_file,MY_SEEK_SET,
1681 &info->aio_result.result))
1682 { /* Skip async io */
1683 my_errno=errno;
1684 DBUG_PRINT("error",("got error: %d, aio_result: %d from aioread, async skipped",
1685 errno, info->aio_result.result.aio_errno));
1686 if (info->request_pos != info->buffer)
1687 {
1688 bmove(info->buffer,info->request_pos,
1689 (size_t) (info->read_end - info->read_pos));
1690 info->request_pos=info->buffer;
1691 info->read_pos-=info->read_length;
1692 info->read_end-=info->read_length;
1693 }
1694 info->read_length=info->buffer_length; /* Use hole buffer */
1695 info->read_function=_my_b_cache_read; /* Use normal IO_READ next */
1696 }
1697 else
1698 info->inited=info->aio_result.pending=1;
1699 }
1700 return 0; /* Block read, async in use */
1701 } /* _my_b_async_read */
1702 #endif
1703
1704
1705 /* Read one byte when buffer is empty */
1706
_my_b_get(IO_CACHE * info)1707 int _my_b_get(IO_CACHE *info)
1708 {
1709 uchar buff;
1710 if ((*(info)->read_function)(info,&buff,1))
1711 return my_b_EOF;
1712 return (int) (uchar) buff;
1713 }
1714
1715 /*
1716 Write a byte buffer to IO_CACHE and flush to disk
1717 if IO_CACHE is full.
1718
1719 RETURN VALUE
1720 1 On error on write
1721 0 On success
1722 -1 On error; my_errno contains error code.
1723 */
1724
_my_b_cache_write(IO_CACHE * info,const uchar * Buffer,size_t Count)1725 int _my_b_cache_write(IO_CACHE *info, const uchar *Buffer, size_t Count)
1726 {
1727 if (Buffer != info->write_buffer)
1728 {
1729 Count= IO_ROUND_DN(Count);
1730 if (!Count)
1731 return 0;
1732 }
1733
1734 if (info->seek_not_done)
1735 {
1736 /*
1737 Whenever a function which operates on IO_CACHE flushes/writes
1738 some part of the IO_CACHE to disk it will set the property
1739 "seek_not_done" to indicate this to other functions operating
1740 on the IO_CACHE.
1741 */
1742 if (mysql_file_seek(info->file, info->pos_in_file, MY_SEEK_SET,
1743 MYF(info->myflags & MY_WME)) == MY_FILEPOS_ERROR)
1744 {
1745 info->error= -1;
1746 return 1;
1747 }
1748 info->seek_not_done=0;
1749 }
1750 if (mysql_file_write(info->file, Buffer, Count, info->myflags | MY_NABP))
1751 return info->error= -1;
1752
1753 info->pos_in_file+= Count;
1754 return 0;
1755 }
1756
1757
1758 /*
1759 In case of a shared I/O cache with a writer we normally do direct
1760 write cache to read cache copy. Simulate this here by direct
1761 caller buffer to read cache copy. Do it after the write so that
1762 the cache readers actions on the flushed part can go in parallel
1763 with the write of the extra stuff. copy_to_read_buffer()
1764 synchronizes writer and readers so that after this call the
1765 readers can act on the extra stuff while the writer can go ahead
1766 and prepare the next output. copy_to_read_buffer() relies on
1767 info->pos_in_file.
1768 */
_my_b_cache_write_r(IO_CACHE * info,const uchar * Buffer,size_t Count)1769 static int _my_b_cache_write_r(IO_CACHE *info, const uchar *Buffer, size_t Count)
1770 {
1771 my_off_t old_pos_in_file= info->pos_in_file;
1772 int res= _my_b_cache_write(info, Buffer, Count);
1773 if (res)
1774 return res;
1775
1776 DBUG_ASSERT(!(info->myflags & MY_ENCRYPT));
1777 DBUG_ASSERT(info->share);
1778 copy_to_read_buffer(info, Buffer, old_pos_in_file);
1779
1780 return 0;
1781 }
1782
1783
1784 /*
1785 Append a block to the write buffer.
1786 This is done with the buffer locked to ensure that we don't read from
1787 the write buffer before we are ready with it.
1788 */
1789
my_b_append(IO_CACHE * info,const uchar * Buffer,size_t Count)1790 int my_b_append(IO_CACHE *info, const uchar *Buffer, size_t Count)
1791 {
1792 size_t rest_length,length;
1793
1794 /*
1795 Assert that we cannot come here with a shared cache. If we do one
1796 day, we might need to add a call to copy_to_read_buffer().
1797 */
1798 DBUG_ASSERT(!info->share);
1799 DBUG_ASSERT(!(info->myflags & MY_ENCRYPT));
1800
1801 lock_append_buffer(info);
1802 rest_length= (size_t) (info->write_end - info->write_pos);
1803 if (Count <= rest_length)
1804 goto end;
1805 memcpy(info->write_pos, Buffer, rest_length);
1806 Buffer+=rest_length;
1807 Count-=rest_length;
1808 info->write_pos+=rest_length;
1809 if (my_b_flush_io_cache(info,0))
1810 {
1811 unlock_append_buffer(info);
1812 return 1;
1813 }
1814 if (Count >= IO_SIZE)
1815 { /* Fill first intern buffer */
1816 length= IO_ROUND_DN(Count);
1817 if (mysql_file_write(info->file,Buffer, length, info->myflags | MY_NABP))
1818 {
1819 unlock_append_buffer(info);
1820 return info->error= -1;
1821 }
1822 Count-=length;
1823 Buffer+=length;
1824 info->end_of_file+=length;
1825 }
1826
1827 end:
1828 memcpy(info->write_pos,Buffer,(size_t) Count);
1829 info->write_pos+=Count;
1830 unlock_append_buffer(info);
1831 return 0;
1832 }
1833
1834
my_b_safe_write(IO_CACHE * info,const uchar * Buffer,size_t Count)1835 int my_b_safe_write(IO_CACHE *info, const uchar *Buffer, size_t Count)
1836 {
1837 /*
1838 Sasha: We are not writing this with the ? operator to avoid hitting
1839 a possible compiler bug. At least gcc 2.95 cannot deal with
1840 several layers of ternary operators that evaluated comma(,) operator
1841 expressions inside - I do have a test case if somebody wants it
1842 */
1843 if (info->type == SEQ_READ_APPEND)
1844 return my_b_append(info, Buffer, Count);
1845 return my_b_write(info, Buffer, Count);
1846 }
1847
1848
1849 /*
1850 Write a block to disk where part of the data may be inside the record
1851 buffer. As all write calls to the data goes through the cache,
1852 we will never get a seek over the end of the buffer
1853 */
1854
my_block_write(IO_CACHE * info,const uchar * Buffer,size_t Count,my_off_t pos)1855 int my_block_write(IO_CACHE *info, const uchar *Buffer, size_t Count,
1856 my_off_t pos)
1857 {
1858 size_t length;
1859 int error=0;
1860
1861 /*
1862 Assert that we cannot come here with a shared cache. If we do one
1863 day, we might need to add a call to copy_to_read_buffer().
1864 */
1865 DBUG_ASSERT(!info->share);
1866 DBUG_ASSERT(!(info->myflags & MY_ENCRYPT));
1867
1868 if (pos < info->pos_in_file)
1869 {
1870 /* Of no overlap, write everything without buffering */
1871 if (pos + Count <= info->pos_in_file)
1872 return (int)mysql_file_pwrite(info->file, Buffer, Count, pos,
1873 info->myflags | MY_NABP);
1874 /* Write the part of the block that is before buffer */
1875 length= (uint) (info->pos_in_file - pos);
1876 if (mysql_file_pwrite(info->file, Buffer, length, pos, info->myflags | MY_NABP))
1877 info->error= error= -1;
1878 Buffer+=length;
1879 pos+= length;
1880 Count-= length;
1881 }
1882
1883 /* Check if we want to write inside the used part of the buffer.*/
1884 length= (size_t) (info->write_end - info->buffer);
1885 if (pos < info->pos_in_file + length)
1886 {
1887 size_t offset= (size_t) (pos - info->pos_in_file);
1888 length-=offset;
1889 if (length > Count)
1890 length=Count;
1891 memcpy(info->buffer+offset, Buffer, length);
1892 Buffer+=length;
1893 Count-= length;
1894 /* Fix length of buffer if the new data was larger */
1895 if (info->buffer+length > info->write_pos)
1896 info->write_pos=info->buffer+length;
1897 if (!Count)
1898 return (error);
1899 }
1900 /* Write at the end of the current buffer; This is the normal case */
1901 if (_my_b_write(info, Buffer, Count))
1902 error= -1;
1903 return error;
1904 }
1905
1906
1907 /* Flush write cache */
1908
1909 #define LOCK_APPEND_BUFFER if (need_append_buffer_lock) \
1910 lock_append_buffer(info);
1911 #define UNLOCK_APPEND_BUFFER if (need_append_buffer_lock) \
1912 unlock_append_buffer(info);
1913
my_b_flush_io_cache(IO_CACHE * info,int need_append_buffer_lock)1914 int my_b_flush_io_cache(IO_CACHE *info, int need_append_buffer_lock)
1915 {
1916 size_t length;
1917 my_bool append_cache= (info->type == SEQ_READ_APPEND);
1918 DBUG_ENTER("my_b_flush_io_cache");
1919 DBUG_PRINT("enter", ("cache: %p", info));
1920
1921 if (!append_cache)
1922 need_append_buffer_lock= 0;
1923
1924 if (info->type == WRITE_CACHE || append_cache)
1925 {
1926 if (info->file == -1)
1927 {
1928 if (real_open_cached_file(info))
1929 DBUG_RETURN((info->error= -1));
1930 }
1931 LOCK_APPEND_BUFFER;
1932
1933 if ((length=(size_t) (info->write_pos - info->write_buffer)))
1934 {
1935 if (append_cache)
1936 {
1937 if (mysql_file_write(info->file, info->write_buffer, length,
1938 info->myflags | MY_NABP))
1939 {
1940 info->error= -1;
1941 DBUG_RETURN(-1);
1942 }
1943 info->end_of_file+= info->write_pos - info->append_read_pos;
1944 info->append_read_pos= info->write_buffer;
1945 DBUG_ASSERT(info->end_of_file == mysql_file_tell(info->file, MYF(0)));
1946 }
1947 else
1948 {
1949 int res= info->write_function(info, info->write_buffer, length);
1950 if (res)
1951 DBUG_RETURN(res);
1952
1953 set_if_bigger(info->end_of_file, info->pos_in_file);
1954 }
1955 info->write_end= (info->write_buffer + info->buffer_length -
1956 ((info->pos_in_file + length) & (IO_SIZE - 1)));
1957 info->write_pos= info->write_buffer;
1958 ++info->disk_writes;
1959 UNLOCK_APPEND_BUFFER;
1960 DBUG_RETURN(info->error);
1961 }
1962 }
1963 #ifdef HAVE_AIOWAIT
1964 else if (info->type != READ_NET)
1965 {
1966 my_aiowait(&info->aio_result); /* Wait for outstanding req */
1967 info->inited=0;
1968 }
1969 #endif
1970 UNLOCK_APPEND_BUFFER;
1971 DBUG_RETURN(0);
1972 }
1973
1974 /*
1975 Free an IO_CACHE object
1976
1977 SYNOPSOS
1978 end_io_cache()
1979 info IO_CACHE Handle to free
1980
1981 NOTES
1982 It's currently safe to call this if one has called init_io_cache()
1983 on the 'info' object, even if init_io_cache() failed.
1984 This function is also safe to call twice with the same handle.
1985 Note that info->file is not reset as the caller may still use ut for my_close()
1986
1987 RETURN
1988 0 ok
1989 # Error
1990 */
1991
end_io_cache(IO_CACHE * info)1992 int end_io_cache(IO_CACHE *info)
1993 {
1994 int error=0;
1995 DBUG_ENTER("end_io_cache");
1996 DBUG_PRINT("enter",("cache: %p", info));
1997
1998 /*
1999 Every thread must call remove_io_thread(). The last one destroys
2000 the share elements.
2001 */
2002 DBUG_ASSERT(!info->share || !info->share->total_threads);
2003
2004 if (info->alloced_buffer)
2005 {
2006 info->alloced_buffer=0;
2007 if (info->file != -1) /* File doesn't exist */
2008 error= my_b_flush_io_cache(info,1);
2009 my_free(info->buffer);
2010 info->buffer=info->read_pos=(uchar*) 0;
2011 }
2012 if (info->type == SEQ_READ_APPEND)
2013 {
2014 /* Destroy allocated mutex */
2015 mysql_mutex_destroy(&info->append_buffer_lock);
2016 }
2017 info->share= 0;
2018 info->type= TYPE_NOT_SET; /* Ensure that flush_io_cache() does nothing */
2019 info->write_end= 0; /* Ensure that my_b_write() fails */
2020 info->write_function= 0; /* my_b_write will crash if used */
2021 DBUG_RETURN(error);
2022 } /* end_io_cache */
2023
2024
2025 /**********************************************************************
2026 Testing of MF_IOCACHE
2027 **********************************************************************/
2028
2029 #ifdef MAIN
2030
2031 #include <my_dir.h>
2032
die(const char * fmt,...)2033 void die(const char* fmt, ...)
2034 {
2035 va_list va_args;
2036 va_start(va_args,fmt);
2037 fprintf(stderr,"Error:");
2038 vfprintf(stderr, fmt,va_args);
2039 fprintf(stderr,", errno=%d\n", errno);
2040 va_end(va_args);
2041 exit(1);
2042 }
2043
open_file(const char * fname,IO_CACHE * info,int cache_size)2044 int open_file(const char* fname, IO_CACHE* info, int cache_size)
2045 {
2046 int fd;
2047 if ((fd=my_open(fname,O_CREAT | O_RDWR,MYF(MY_WME))) < 0)
2048 die("Could not open %s", fname);
2049 if (init_io_cache(info, fd, cache_size, SEQ_READ_APPEND, 0,0,MYF(MY_WME)))
2050 die("failed in init_io_cache()");
2051 return fd;
2052 }
2053
close_file(IO_CACHE * info)2054 void close_file(IO_CACHE* info)
2055 {
2056 end_io_cache(info);
2057 my_close(info->file, MYF(MY_WME));
2058 }
2059
main(int argc,char ** argv)2060 int main(int argc, char** argv)
2061 {
2062 IO_CACHE sra_cache; /* SEQ_READ_APPEND */
2063 MY_STAT status;
2064 const char* fname="/tmp/iocache.test";
2065 int cache_size=16384;
2066 char llstr_buf[22];
2067 int max_block,total_bytes=0;
2068 int i,num_loops=100,error=0;
2069 char *p;
2070 char* block, *block_end;
2071 MY_INIT(argv[0]);
2072 max_block = cache_size*3;
2073 if (!(block=(char*)my_malloc(max_block,MYF(MY_WME))))
2074 die("Not enough memory to allocate test block");
2075 block_end = block + max_block;
2076 for (p = block,i=0; p < block_end;i++)
2077 {
2078 *p++ = (char)i;
2079 }
2080 if (my_stat(fname,&status, MYF(0)) &&
2081 my_delete(fname,MYF(MY_WME)))
2082 {
2083 die("Delete of %s failed, aborting", fname);
2084 }
2085 open_file(fname,&sra_cache, cache_size);
2086 for (i = 0; i < num_loops; i++)
2087 {
2088 char buf[4];
2089 int block_size = abs(rand() % max_block);
2090 int4store(buf, block_size);
2091 if (my_b_append(&sra_cache,buf,4) ||
2092 my_b_append(&sra_cache, block, block_size))
2093 die("write failed");
2094 total_bytes += 4+block_size;
2095 }
2096 close_file(&sra_cache);
2097 my_free(block);
2098 if (!my_stat(fname,&status,MYF(MY_WME)))
2099 die("%s failed to stat, but I had just closed it,\
2100 wonder how that happened");
2101 printf("Final size of %s is %s, wrote %d bytes\n",fname,
2102 llstr(status.st_size,llstr_buf),
2103 total_bytes);
2104 my_delete(fname, MYF(MY_WME));
2105 /* check correctness of tests */
2106 if (total_bytes != status.st_size)
2107 {
2108 fprintf(stderr,"Not the same number of bytes actually in file as bytes \
2109 supposedly written\n");
2110 error=1;
2111 }
2112 exit(error);
2113 return 0;
2114 }
2115 #endif
2116