redis源码阅读之aof

点击( 此处 )折叠或打开

1. #
2. # If unsure , use "everysec" .
4. # appendfsync always
5. appendfsync everysec
6. # appendfsync no

点击( 此处 )折叠或打开

1. / * Called when the user switches from "appendonly no" to "appendonly yes"
2.   * at runtime using the CONFIG command . * /
3. int startAppendOnly ( void ) {
4.     char cwd [ MAXPATHLEN ] ; / * Current working dir path for error messages . * /
5.      int newfd ;
7.     newfd = open ( server . aof_filename , O_WRONLY | O_APPEND | O_CREAT , 0644 ) ;
8.     serverAssert ( server . aof_state = = AOF_OFF ) ;
9.      if ( newfd = = - 1 ) {
10.         char * cwdp = getcwd ( cwd , MAXPATHLEN ) ;
12.         serverLog ( LL_WARNING ,
13.              "Redis needs to enable the AOF but can't open the "
14.              "append only file %s (in server root dir %s): %s" ,
15.             server . aof_filename ,
16.             cwdp ? cwdp : "unknown" ,
17.             strerror ( errno ) ) ;
18.         return C_ERR ;
19.      }
20.      if ( server . rdb_child_pid ! = - 1 ) {
21.         server . aof_rewrite_scheduled = 1 ;
22.         serverLog ( LL_WARNING , "AOF was enabled but there is already a child process saving an RDB file on disk. An AOF background was scheduled to start when possible." ) ;
23.      } else {
24.          / * If there is a pending AOF rewrite , we need to switch it off and
25.           * start a new one : the old one cannot be reused becuase it is not
26.           * accumulating the AOF buffer . * /
27.          if ( server . aof_child_pid ! = - 1 ) {
28.             serverLog ( LL_WARNING , "AOF was enabled but there is already an AOF rewriting in background. Stopping background AOF and starting a rewrite now." ) ;
29.             killAppendOnlyChild ( ) ;
30.          }
31.          if ( rewriteAppendOnlyFileBackground ( ) = = C_ERR ) {
32.             close ( newfd ) ;
33.             serverLog ( LL_WARNING , "Redis needs to enable the AOF but can't trigger a background AOF rewrite operation. Check the above logs for more info about the error." ) ;
34.             return C_ERR ;
35.          }
36.      }
37.      / * We correctly switched on AOF , now wait for the rewrite to be complete
38.       * in order to append data on disk . * /
39.     server . aof_state = AOF_WAIT_REWRITE ;
40.     server . aof_last_fsync = server . unixtime ;
41.     server . aof_fd = newfd ;
42.     return C_OK ;
43. }

点击( 此处 )折叠或打开

1. ssize_t aofWrite ( int fd , const char * buf , size_t len ) {
2.     ssize_t nwritten = 0 , totwritten = 0 ;
4.      while ( len ) {
5.         nwritten = write ( fd , buf , len ) ;
7.          if ( nwritten < 0 ) {
8.              if ( errno = = EINTR ) {
9.                 continue ;
10.              }
11.             return totwritten ? totwritten : - 1 ;
12.          }
14.          len - = nwritten ;
15.         buf + = nwritten ;
16.         totwritten + = nwritten ;
17.      }
19.     return totwritten ;
20. }

点击( 此处 )折叠或打开

1. #define AOF_WRITE_LOG_ERROR_RATE 30 / * Seconds between errors logging . * /
2. void flushAppendOnlyFile ( int force ) {
3.     ssize_t nwritten ;
4.      int sync_in_progress = 0 ;
5.     mstime_t latency ;
7.      if ( sdslen ( server . aof_buf ) = = 0 ) return ;
9.      if ( server . aof_fsync = = AOF_FSYNC_EVERYSEC )
10.         sync_in_progress = bioPendingJobsOfType ( BIO_AOF_FSYNC ) ! = 0 ;//bio有讲过
12.      if ( server . aof_fsync = = AOF_FSYNC_EVERYSEC & & ! force ) {
13.          / * With this append fsync policy we do background fsyncing .
14.           * If the fsync is still in progress we can try to delay
15.           * the write for a couple of seconds . * /
16.          if ( sync_in_progress ) {
17.              if ( server . aof_flush_postponed_start = = 0 ) {
18.                  / * No previous write postponing , remember that we are
19.                   * postponing the flush and return . * /
20.                 server . aof_flush_postponed_start = server . unixtime ;
21.                 return ;
22.              } else if ( server . unixtime - server . aof_flush_postponed_start < 2 ) {
23.                  / * We were already waiting for fsync to finish , but for less
24.                   * than two seconds this is still ok . Postpone again . * /
25.                 return ;
26.              }
27.              / * Otherwise fall trough , and go write since we can ' t wait
28.               * over two seconds . * /
29.             server . aof_delayed_fsync + + ;
30.             serverLog ( LL_NOTICE , "Asynchronous AOF fsync is taking too long (disk is busy?). Writing the AOF buffer without waiting for fsync to complete, this may slow down Redis." ) ;
31.          }
32.      }
33.      / * We want to perform a single write . This should be guaranteed atomic
34.       * at least if the filesystem we are writing is a real physical one .
35.       * While this will save us against the server being killed I don ' t think
36.       * there is much to do about the whole server stopping for power problems
37.       * or alike * /
39.     latencyStartMonitor ( latency ) ;
40.     nwritten = aofWrite ( server . aof_fd , server . aof_buf , sdslen ( server . aof_buf ) ) ;
41.     latencyEndMonitor ( latency ) ;
42.      / * We want to capture different events for delayed writes :
43.       * when the delay happens with a pending fsync , or with a saving child
44.       * active , and when the above two conditions are missing .
45.       * We also use an additional event name to save all samples which is
46.       * useful for graphing / monitoring purposes . * /
47.      if ( sync_in_progress ) {
48.         latencyAddSampleIfNeeded ( "aof-write-pending-fsync" , latency ) ;
49.      } else if ( server . aof_child_pid ! = - 1 | | server . rdb_child_pid ! = - 1 ) {
50.         latencyAddSampleIfNeeded ( "aof-write-active-child" , latency ) ;
51.      } else {
52.         latencyAddSampleIfNeeded ( "aof-write-alone" , latency ) ;
53.      }
54.     latencyAddSampleIfNeeded ( "aof-write" , latency ) ;
56.      / * We performed the write so reset the postponed flush sentinel to zero . * /
57.     server . aof_flush_postponed_start = 0 ;
59.      if ( nwritten ! = ( ssize_t ) sdslen ( server . aof_buf ) ) {
60.         static time_t last_write_error_log = 0 ;
61.          int can_log = 0 ;
63.          / * Limit logging rate to 1 line per AOF_WRITE_LOG_ERROR_RATE seconds . * /
64.          if ( ( server . unixtime - last_write_error_log ) > AOF_WRITE_LOG_ERROR_RATE ) {
65.             can_log = 1 ;
66.             last_write_error_log = server . unixtime ;
67.          }
69.          / * Log the AOF write error and record the error code . * /
70.          if ( nwritten = = - 1 ) {
71.              if ( can_log ) {
72.                 serverLog ( LL_WARNING , "Error writing to the AOF file: %s" ,
73.                     strerror ( errno ) ) ;
74.                 server . aof_last_write_errno = errno ;
75.              }
76.          } else {
77.              if ( can_log ) {
78.                 serverLog ( LL_WARNING , "Short write while writing to "
79.                                         "the AOF file: (nwritten=%lld, "
80.                                         "expected=%lld)" ,
81.                                         ( long long ) nwritten ,
82.                                         ( long long ) sdslen ( server . aof_buf ) ) ;
83.              }
85.              if ( ftruncate ( server . aof_fd , server . aof_current_size ) = = - 1 ) {
86.                  if ( can_log ) {
87.                     serverLog ( LL_WARNING , "Could not remove short write "
88.                               "from the append-only file. Redis may refuse "
89.                               "to load the AOF the next time it starts. "
90.                               "ftruncate: %s" , strerror ( errno ) ) ;
91.                  }
92.              } else {
93.                  / * If the ftruncate ( ) succeeded we can set nwritten to
94.                   * - 1 since there is no longer partial data into the AOF . * /
95.                 nwritten = - 1 ;
96.              }
97.             server . aof_last_write_errno = ENOSPC ;
98.          }
100.          / * Handle the AOF write error . * /
101.          if ( server . aof_fsync = = AOF_FSYNC_ALWAYS ) {
102.              / * We can ' t recover when the fsync policy is ALWAYS since the
103.               * reply for the client is already in the output buffers , and we
104.               * have the contract with the user that on acknowledged write data
105.               * is synced on disk . * /
106.             serverLog ( LL_WARNING , "Can't recover from AOF write error when the AOF fsync policy is 'always'. Exiting..." ) ;
107.              exit ( 1 ) ;
108.          } else {
109.              / * Recover from failed write leaving data into the buffer . However
110.               * set an error to stop accepting writes as long as the error
111.               * condition is not cleared . * /
112.             server . aof_last_write_status = C_ERR ;
114.              / * Trim the sds buffer if there was a partial write , and there
115.               * was no way to undo it with ftruncate ( 2 ) . * /
116.              if ( nwritten > 0 ) {
117.                 server . aof_current_size + = nwritten ;
118.                 sdsrange ( server . aof_buf , nwritten , - 1 ) ;
119.              }
120.             return ; / * We ' ll try again on the next call . . . * /
121.          }
122.      } else {
123.          / * Successful write ( 2 ) . If AOF was in error state , restore the
124.           * OK state and log the event . * /
125.          if ( server . aof_last_write_status = = C_ERR ) {
126.             serverLog ( LL_WARNING ,
127.                  "AOF write error looks solved, Redis can write again." ) ;
128.             server . aof_last_write_status = C_OK ;
129.          }
130.      }
131.     server . aof_current_size + = nwritten ;
133.      / * Re - use AOF buffer when it is small enough . The maximum comes from the
134.       * arena size of 4k minus some overhead ( but is otherwise arbitrary ) . * /
135.      if ( ( sdslen ( server . aof_buf ) + sdsavail ( server . aof_buf ) ) < 4000 ) {
136.         sdsclear ( server . aof_buf ) ;
137.      } else {
138.         sdsfree ( server . aof_buf ) ;
139.         server . aof_buf = sdsempty ( ) ;
140.      }
142.      / * Don ' t fsync if no - appendfsync - on - rewrite is set to yes and there are
143.       * children doing I / O in the background . * /
144.      if ( server . aof_no_fsync_on_rewrite & &
145.          ( server . aof_child_pid ! = - 1 | | server . rdb_child_pid ! = - 1 ) )
146.             return ;
148.      / * Perform the fsync if needed . * /
149.      if ( server . aof_fsync = = AOF_FSYNC_ALWAYS ) {
150.          / * aof_fsync is defined as fdatasync ( ) for Linux in order to avoid
151.           * flushing metadata . * /
152.         latencyStartMonitor ( latency ) ;
153.         aof_fsync ( server . aof_fd ) ; / * Let ' s try to get this data on the disk * /
154.         latencyEndMonitor ( latency ) ;
155.         latencyAddSampleIfNeeded ( "aof-fsync-always" , latency ) ;
156.         server . aof_last_fsync = server . unixtime ;
157.      } else if ( ( server . aof_fsync = = AOF_FSYNC_EVERYSEC & &
158.                 server . unixtime > server . aof_last_fsync ) ) {
159.          if ( ! sync_in_progress ) aof_background_fsync ( server . aof_fd ) ;
160.         server . aof_last_fsync = server . unixtime ;
161.      }
162. }

点击( 此处 )折叠或打开

1. #define AOF_RW_BUF_BLOCK_SIZE ( 1024 * 1024 * 10 ) / * 10 MB per block * /
3. typedef struct aofrwblock {
4.     unsigned long used , free ;
5.     char buf [ AOF_RW_BUF_BLOCK_SIZE ] ;
6. } aofrwblock ;

点击( 此处 )折叠或打开

1. / * Append data to the AOF rewrite buffer , allocating new blocks if needed . * /
2. void aofRewriteBufferAppend ( unsigned char * s , unsigned long len ) {
3.     listNode * ln = listLast ( server . aof_rewrite_buf_blocks ) ;
4.     aofrwblock * block = ln ? ln - > value : NULL ;
6.      while ( len ) {
7.          / * If we already got at least an allocated block , try appending
8.           * at least some piece into it . * /
9.          if ( block ) {
10.             unsigned long thislen = ( block - > free < len ) ? block -