nl_tokens: This function drives the writeback of dirty files .
pl_tokens: function _writeback ( immediate_file , cb ) { assert . optionalString ( immediate_file , 'immediate_file' ) ; assert . optionalFunc ( cb , 'callback' ) ; var log = this . log ; var self = this ; function _doUpload ( _arg , _cb ) { var cachefile = path . join ( self . location , 'fscache' , _arg . cachefile ) ; log . debug ( 'writeback %s => %s' , _arg . cachefile , _arg . fname ) ; fs . stat ( cachefile , function onStatDone ( s_err , stats ) { if ( s_err ) { // This should never happen log . error ( s_err , '_doUpload(%s): error stat-ing local ' + 'file (%s)' , _arg . fname , cachefile ) ; _cb ( s_err , _arg ) ; return ; } // Only read the current size in case the file gets extended // while we're uploading. In that case it will be dirty again // and be re-uploaded. var rd_size = stats . size ; if ( rd_size > 0 ) rd_size -= 1 ; var r_opts = { start : 0 , end : rd_size } ; var m_opts = { size : stats . size } ; var rstream = fs . createReadStream ( cachefile , r_opts ) ; rstream . once ( 'error' , function onFileError ( err ) { log . error ( err , '_doUpload(%s): error reading from local ' + 'file (%s)' , _arg . fname , cachefile ) ; _cb ( <mask> , _arg ) ; } ) ; rstream . once ( 'open' , function onOpen ( ) { self . manta . put ( manta . path ( _arg . fname , true ) , rstream , m_opts , function ( err ) { if ( err ) { log . error ( err , '_doUpload(%s): error pushing (%s) ' + 'to manta:' , _arg . fname , cachefile ) ; _cb ( err , _arg ) ; } else { log . debug ( '_doUpload(%s) pushed to (%s):' , cachefile , _arg . fname ) ; _cb ( null , _arg ) ; } } ) ; } ) ; } ) ; } function _fileUploaded ( _arg , _cb ) { if ( ! wb_pending [ _arg . fname ] ) { // if no more writes while we were uploading, we remove it from // the DB since its no longer dirty var key = sprintf ( DIRTY_KEY_FMT , _arg . fname ) ; self . db . batch ( ) . del ( key ) . write ( function onBatchDel ( err ) { if ( err ) { log . warn ( err , 'dirty del(%s): failed' , _arg . fname ) ; _cb ( errno ( err , _arg ) ) ; } else { log . trace ( 'dirty del(%s): done' , _arg . fname ) ; _cb ( null , _arg ) ; } } ) ; } } function _updateLastMod ( _arg , _cb ) { // Update the last modified time for this file into the cache. // Ignore errors here since this can only impact our view of when the // file is stale. self . get ( _arg . fname , function ( err , info ) { if ( err ) { _cb ( null , _arg ) ; return ; } var upd = new Date ( ) ; info . headers [ 'last-modified' ] = upd . toUTCString ( ) ; info . last_stat = upd . getTime ( ) ; self . put ( _arg . fname , info , function ( err2 ) { _cb ( null , _arg ) ; } ) ; } ) ; } function _wb_done ( err , res ) { // just use the return from the first operation to get the args var args = res . operations [ 0 ] . result ; if ( err ) { log . error ( err , 'writeback: error %s' , args . fname ) ; // XXX JJ generalize error handling // e.g. if (err.code === 'UploadTimeout') ... // other manta errors, etc. // On upload timeout we won't have called the _fileUploaded // function in the pipeline, so we still have the file in the DB, // but not the wb_pending list. We'll add this entry back into the // in-memory wb_pending list (or overwrite the existing entry if // the file was updated while we tried to upload), but with a // timestamp one minute in the future, so that we'll retry the // upload of this file a little later. We still keep trying to // upload any other files in the list. This handles the case in // which the failure was somehow relevant only to this file. If all // uploads fail, we'll have just added them back to the writeback // list for future retries. var plusminute = new Date ( ) . getTime ( ) ; plusminute += 60000 ; wb_pending [ args . fname ] = { cachefile : args . cachefile , timestamp : plusminute } ; } else { log . debug ( 'writeback %s done' , args . fname ) ; } var pending_op = wb_inprogress [ args . fname ] . op ; var pending_cb = wb_inprogress [ args . fname ] . cb ; var pending_info = wb_inprogress [ args . fname ] . info ; // remove it from the in-memory inprogress list delete wb_inprogress [ args . fname ] ; wb_num -- ; if ( pending_op ) { log . debug ( 'writeback %s, pending %s' , args . fname , pending_op ) ; // We had to block an operation while the upload was in progress. // Perform that operation now and schedule _writeback to continue. setTimeout ( _writeback . bind ( self ) , 1 ) ; if ( pending_op === 'del' ) { self . del ( args . fname , pending_cb ) ; } else if ( pending_op === 'trunc' ) { self . truncate ( args . fname , pending_info , pending_cb ) ; } else if ( pending_op === 'cb' ) { pending_cb ( ) ; } else { log . error ( 'writeback %s, pending %s unknown' , args . fname , pending_op ) ; } } else { // this is the normal path, just take another lap var wb = _writeback . bind ( self ) ; wb ( ) ; } } if ( immediate_file ) { if ( wb_pending [ immediate_file ] ) { // the file is dirty, upload it now log . debug ( '_writeback immd. %s: pending' , immediate_file ) ; file_data = { fname : immediate_file , cachefile : wb_pending [ immediate_file ] . cachefile } ; // remove it from the in-memory list delete wb_pending [ immediate_file ] ; // add it to the in-memory in progress list with cb pending wb_inprogress [ immediate_file ] = { op : 'cb' , cb : cb } ; wb_num ++ ; // here is where we setup the writeback pipeline vasync . pipeline ( { funcs : [ _doUpload , _fileUploaded , _updateLastMod ] , arg : file_data } , _wb_done ) ; } else if ( wb_inprogress [ immediate_file ] ) { // the file is already being uploaded now, setup cb if ( wb_inprogress [ immediate_file ] . op ) { // there shouldn't be any pending op in this case log . error ( 'writeback immd. %s: already pending %s' , immediate_file , wb_inprogress [ immediate_file ] . op ) ; } log . debug ( '_writeback immd. %s: in progress' , immediate_file ) ; wb_inprogress [ immediate_file ] = { op : 'cb' , cb : cb } ; } else { // the file is not dirty and not being uploaded, just run cb log . debug ( '_writeback immd. %s: do nothing' , immediate_file ) ; cb ( null ) ; } return ; } var now = new Date ( ) . getTime ( ) ; var threshold = now - self . wbtime ; // When we iterate using 'in', the elements in wb_pending are probably // ordered by the order in which they were defined. On startup, when // pulling out of the DB, we define them in the order we get them from the // DB which, since its a btree, will be alphabetical. Thus, we have to look // at each entry to see if its time to write it back. That is, the entries // might not be in timestamp order so we can't stop when we see an entry // that's not ready to be written back. // // Once we start processing the list, we'll get called back and keep making // a pass over the list until we have made a full pass with no writeback. // At that time, if there are still elements in the list, we schedule // ourselves to wakeup later and process the list again. var wb_running = false ; var more_files = false ; var fname ; var file_data = { } ; for ( fname in wb_pending ) { if ( threshold > wb_pending [ fname ] . timestamp ) { if ( wb_inprogress [ fname ] ) { // This file is still in the middle of an upload and its dirty // again and time to writeback again. We skip this file until // the current upload finishes and we come back around. continue ; } file_data = { fname : fname , cachefile : wb_pending [ fname ] . cachefile } ; // remove it from the in-memory list delete wb_pending [ fname ] ; // add it to the in-memory in progress list with nothing pending wb_inprogress [ fname ] = { } ; wb_num ++ ; wb_running = true ; // here is where we setup the writeback pipeline vasync . pipeline ( { funcs : [ _doUpload , _fileUploaded , _updateLastMod ] , arg : file_data } , _wb_done ) ; if ( wb_num >= self . num_par ) { // we've hit our limit for parallel writebacks, break out of // the loop since we'll callback to _writeback when one of the // uploads has finished break ; } } else { more_files = true ; } } if ( ! wb_running ) { if ( more_files ) { setTimeout ( _writeback . bind ( self ) , self . wbtime ) ; } else { writeback_enabled = false ; } } }