redis源码阅读之rio

点击( 此处 )折叠或打开

1. struct _rio {
2.      / * Backend functions .
3.       * Since this functions do not tolerate short writes or reads the return
4.       * value is simplified to : zero on error , non zero on complete success . * /
5.     size_t ( * read ) ( struct _rio * , void * buf , size_t len ) ;
6.     size_t ( * write ) ( struct _rio * , const void * buf , size_t len ) ;
7.     off_t ( * tell ) ( struct _rio * ) ;
8.      int ( * flush ) ( struct _rio * ) ;
9.      / * The update_cksum method if not NULL is used to compute the checksum of
10.       * all the data that was read or written so far . The method should be
11.       * designed so that can be called with the current checksum , and the buf
12.       * and len fields pointing to the new block of data to add to the checksum
13.       * computation . * /
14.     void ( * update_cksum ) ( struct _rio * , const void * buf , size_t len ) ;
16.      / * The current checksum * /
17.     uint64_t cksum ;
19.      / * number of bytes read or written * /
20.     size_t processed_bytes ;
22.      / * maximum single read or write chunk size * /
23.     size_t max_processing_chunk ;
25.      / * Backend - specific vars . * /
26.     union {
27.          / * In - memory buffer target . * /
28.         struct {
29.             sds ptr ;
30.             off_t pos ;
31.          } buffer ;
32.          / * Stdio file pointer target . * /
33.         struct {
34.             FILE * fp ;
35.             off_t buffered ; / * Bytes written since last fsync . * /
36.             off_t autosync ; / * fsync after ' autosync ' bytes written . * /
37.          } file ;
38.          / * Multiple FDs target ( used to write to N sockets ) . * /
39.         struct {
40.              int * fds ; / * File descriptors . * /
41.              int * state ; / * Error state of each fd . 0 ( if ok ) or errno . * /
42.              int numfds ;
43.             off_t pos ;
44.             sds buf ;
45.          } fdset ;
46.      } io ;
47. } ;
49. typedef struct _rio rio ;

点击( 此处 )折叠或打开

1. static inline size_t rioWrite ( rio * r , const void * buf , size_t len ) {
2.      while ( len ) {
3.         size_t bytes_to_write = ( r - > max_processing_chunk & & r - > max_processing_chunk < len ) ? r - > max_processing_chunk : len ;
4.          if ( r - > update_cksum ) r - > update_cksum ( r , buf , bytes_to_write ) ;
5.          if ( r - > write ( r , buf , bytes_to_write ) = = 0 ) / * 0表示失败；非0表示成功 * /
6.             return 0 ;
7.         buf = ( char * ) buf + bytes_to_write ;
8.          len - = bytes_to_write ;
9.         r - > processed_bytes + = bytes_to_write ;
10.      }
11.     return 1 ;
12. }
14. static inline size_t rioRead ( rio * r , void * buf , size_t len ) {
15.      while ( len ) {
16.         size_t bytes_to_read = ( r - > max_processing_chunk & & r - > max_processing_chunk < len ) ? r - > max_processing_chunk : len ;
17.          if ( r - > read ( r , buf , bytes_to_read ) = = 0 ) / * 0表示失败；非0表示成功 * /
18.             return 0 ;
19.          if ( r - > update_cksum ) r - > update_cksum ( r , buf , bytes_to_read ) ;
20.         buf = ( char * ) buf + bytes_to_read ;
21.          len - = bytes_to_read ;
22.         r - > processed_bytes + = bytes_to_read ;
23.      }
24.     return 1 ;
25. }
27. static inline off_t rioTell ( rio * r ) {
28.     return r - > tell ( r ) ;
29. }
31. static inline int rioFlush ( rio * r ) {
32.     return r - > flush ( r ) ;
33. }

点击( 此处 )折叠或打开

1. / * Returns 1 or 0 for success / failure . * /
2. static size_t rioBufferWrite ( rio * r , const void * buf , size_t len ) {
3.     r - > io . buffer . ptr = sdscatlen ( r - > io . buffer . ptr , ( char * ) buf , len ) ;
4.     r - > io . buffer . pos + = len ;
5.     return 1 ;
6. }
8. / * Returns 1 or 0 for success / failure . * /
9. static size_t rioBufferRead ( rio * r , void * buf , size_t len ) {
10.      if ( sdslen ( r - > io . buffer . ptr ) - r - > io . buffer . pos < len )
11.         return 0 ; / * not enough buffer to return len bytes . * /
12.     memcpy ( buf , r - > io . buffer . ptr + r - > io . buffer . pos , len ) ;
13.     r - > io . buffer . pos + = len ;
14.     return 1 ;
15. }
17. / * Returns read / write position in buffer . * /
18. static off_t rioBufferTell ( rio * r ) {
19.     return r - > io . buffer . pos ;
20. }
22. / * Flushes any buffer to target device if applicable . Returns 1 on success
23.   * and 0 on failures . * /
24. static int rioBufferFlush ( rio * r ) {
25.     UNUSED ( r ) ;
26.     return 1 ; / * Nothing to do , our write just appends to the buffer . * /
27. }
29. static const rio rioBufferIO = {
30.     rioBufferRead ,
31.     rioBufferWrite ,
32.     rioBufferTell ,
33.     rioBufferFlush ,
34.      NULL , / * update_checksum * /
35.     0 , / * current checksum * /
36.     0 , / * bytes read or written * /
37.     0 , / * read / write chunk size * /
38.      { { NULL , 0 } } / * union for io - specific vars * /
39. } ;
41. void rioInitWithBuffer ( rio * r , sds s ) {
42.      * r = rioBufferIO ;
43.     r - > io . buffer . ptr = s ;
44.     r - > io . buffer . pos = 0 ;
45. }

点击( 此处 )折叠或打开

1. / * Returns 1 or 0 for success / failure . * /
2. static size_t rioFileWrite ( rio * r , const void * buf , size_t len ) {
3.     size_t retval ;
5.     retval = fwrite ( buf , len , 1 , r - > io . file . fp ) ;
6.     r - > io . file . buffered + = len ;
8.      if ( r - > io . file . autosync & &
9.         r - > io . file . buffered > = r - > io . file . autosync )
10.      {
11.         fflush ( r - > io . file . fp ) ;
12.         aof_fsync ( fileno ( r - > io . file . fp ) ) ;
13.         r - > io . file . buffered = 0 ;
14.      }
15.     return retval ;
16. }
18. / * Returns 1 or 0 for success / failure . * /
19. static size_t rioFileRead ( rio * r , void * buf , size_t len ) {
20.     return fread ( buf , len , 1 , r - > io . file . fp ) ;
21. }
23. / * Returns read / write position in file . * /
24. static off_t rioFileTell ( rio * r ) {
25.     return ftello ( r - > io . file . fp ) ;
26. }
28. / * Flushes any buffer to target device if applicable . Returns 1 on success
29.   * and 0 on failures . * /
30. static int rioFileFlush ( rio * r ) {
31.     return ( fflush ( r - > io . file . fp ) = = 0 ) ? 1 : 0 ;
32. }
34. static const rio rioFileIO = {
35.     rioFileRead ,
36.     rioFileWrite ,
37.     rioFileTell ,
38.     rioFileFlush ,
39.      NULL , / * update_checksum * /
40.     0 , / * current checksum * /
41.     0 , / * bytes read or written * /
42.     0 , / * read / write chunk size * /
43.      { { NULL , 0 } } / * union for io - specific vars * /
44. } ;
46. void rioInitWithFile ( rio * r , FILE * fp ) {
47.      * r = rioFileIO ;
48.     r - > io . file . fp = fp ;
49.     r - > io . file . buffered = 0 ;
50.     r - > io . file . autosync = 0 ;
51. }

点击( 此处 )折叠或打开

1. / * Returns 1 or 0 for success / failure .
2.   * The function returns success as long as we are able to correctly write
3.   * to at least one file descriptor .
4.   *
5.   * When buf is NULL and len is 0 , the function performs a flush operation
6.   * if there is some pending buffer , so this function is also used in order
7.   * to implement rioFdsetFlush ( ) . * /
8. static size_t rioFdsetWrite ( rio * r , const void * buf , size_t len ) {
9.     ssize_t retval ;
10.      int j ;
11.     unsigned char * p = ( unsigned char * ) buf ;
12.      int doflush = ( buf = = NULL & & len = = 0 ) ;
14.      / * To start we always append to our buffer . If it gets larger than
15.       * a given size , we actually write to the sockets . * /
16.      if ( len ) {
17.         r - > io . fdset . buf = sdscatlen ( r - > io . fdset . buf , buf , len ) ;
18.          len = 0 ; / * Prevent entering the while below if we don ' t flush . * /
19.          if ( sdslen ( r - > io . fdset . buf ) > PROTO_IOBUF_LEN ) doflush = 1 ;
20.      }
22.      if ( doflush ) {
23.         p = ( unsigned char * ) r - > io . fdset . buf ;
24.          len = sdslen ( r - > io . fdset . buf ) ;
25.      }
27.      / * Write in little chunchs so that when there are big writes we
28.       * parallelize while the kernel is sending data in background to
29.       * the TCP socket . * /
30.      while ( len ) {
31.         size_t count = len < 1024 ? len : 1024 ;
32.          int broken = 0 ;
33.          for ( j = 0 ; j < r - > io . fdset . numfds ; j + + ) {
34.              if ( r - > io . fdset . state [ j ] ! = 0 ) {
35.                  / * Skip FDs alraedy in error . * /
36.                 broken + + ;
37.                 continue ;
38.              }
40.              / * Make sure to write ' count ' bytes to the socket regardless
41.               * of short writes . * /
42.             size_t nwritten = 0 ;
43.              while ( nwritten ! = count ) {
44.                 retval = write ( r - > io . fdset . fds [ j ] , p + nwritten , count - nwritten ) ;
45.                  if ( retval < = 0 ) {
46.                      / * With blocking sockets , which is the sole user of this
47.                       * rio target , EWOULDBLOCK is returned only because of
48.                       * the SO_SNDTIMEO socket option , so we translate the error
49.                       * into one more recognizable by the user . * /
50.                      if ( retval = = - 1 & & errno = = EWOULDBLOCK ) errno = ETIMEDOUT ;
51.                     break ;
52.                  }
53.                 nwritten + = retval ;
54.              }
56.              if ( nwritten ! = count ) {
57.                  / * Mark this FD as broken . * /
58.                 r - > io . fdset . state [ j ] = errno ;
59.                  if ( r - > io . fdset . state [ j ] = = 0 ) r - > io . fdset . state [ j ] = EIO ;
60.              }
61.          }
62.          if ( broken = = r - > io . fdset . numfds ) return 0 ; / * All the FDs in error . * /
63.         p + = count ;
64.