text:
/**
 * @file libsyndicate/download.cpp
 * @author Jude Nelson
 * @date 9 Mar 2016
 *
 * @brief Functions to support download capabilities
 *
 * @see libsyndicate/download.h
 */

#include "libsyndicate/download.h"

/**
 * @brief Download contexts set
 */
struct md_download_set {

   md_pending_set_t* waiting;           ///< pointers to download contexts for which we are waiting

   sem_t sem;                           ///< block on this until at least one of waiting has been finalized
};

/**
 * @brief Download context
 */
struct md_download_context {

   struct md_bound_response_buffer brb; ///< A bounded response buffer

   CURL* curl;                    ///< Ptr to a CURL object

   int curl_rc;                   ///< Stores CURL error code
   int http_status;               ///< Stores HTTP status
   int transfer_errno;            ///< Stores CURL-reported system errno, if an error occurred
   volatile bool cancelled;       ///< If true, this was cancelled
   char* effective_url;           ///< Stores final URL that resolved to data

   volatile bool initialized;     ///< If true, then this download context has been initialized
   volatile bool pending;         ///< If true, then this download context is in the process of being started
   volatile bool cancelling;      ///< If true, then this download context is in the process of being cancelled
   volatile bool running;         ///< If true, then this download is enqueued on the downloader
   volatile bool finalized;       ///< If true, then this download has finished
   int ref_count;                 ///< Number of threads referencing this download

   pthread_mutex_t finalize_lock; ///< Lock to serialize operations that change the above flags (primarily related to finalization)

   struct md_download_set* dlset; ///< Parent group containing this context

   sem_t sem;                     ///< Client holds this to be woken up when the download finishes

   void* cls;                     ///< Associated download state
};

/**
 * @brief Download worker
 *
 * Container for download information
 */
struct md_downloader {

   char* name;                          ///< Downloader name
   pthread_t thread;                    ///< CURL thread for downloading

   md_downloading_map_t* downloading;   ///< Currently-running downloads
   pthread_rwlock_t downloading_lock;   ///< Guards downloading and curlm

   md_pending_set_t* pending;           ///< To be inserted into the downloading map
   pthread_rwlock_t pending_lock;       ///< Guards pending
   volatile bool has_pending;           ///< Flag that download is pending

   md_pending_set_t* cancelling;        ///< To be removed from the downloading map
   pthread_rwlock_t cancelling_lock;    ///< Guards cancelling_lock
   volatile bool has_cancelling;        ///< Flag that a download is cancelling

   CURLM* curlm;                        ///< Multi-download

   bool running;                        ///< If true, then this downloader is running
   bool inited;                         ///< If true, then this downloader is fully initialized
};

/**
 * @brief Download loop structure
 *
 * Container for downloads
 */
struct md_download_loop {

   struct md_downloader* dl;               ///< Downloader

   struct md_download_context** downloads; ///< Downloads
   int num_downloads;                      ///< Number of downloads

   struct md_download_set dlset;           ///< A download set with download contexts for which we are waiting

   bool started;                           ///< Flag whether the download loop is in progress
};

static void* md_downloader_main( void* arg );
int md_downloader_finalize_download_context( struct md_download_context* dlctx, int curl_rc );

// download context sets (like an FDSET)
int md_download_set_init( struct md_download_set* dlset );
int md_download_set_free( struct md_download_set* dlset );
int md_download_set_add( struct md_download_set* dlset, struct md_download_context* dlctx );
int md_download_set_clear( struct md_download_set* dlset, struct md_download_context* dlctx );    // don't use inside a e.g. for() loop where you're iterating over a download set
int md_download_set_wakeup( struct md_download_set* dlset );
size_t md_download_set_size( struct md_download_set* dlset );
bool md_download_set_contains( struct md_download_set* dlset, struct md_download_context* dlctx );

// iterating through waiting
md_download_set_iterator md_download_set_begin( struct md_download_set* dlset );
md_download_set_iterator md_download_set_end( struct md_download_set* dlset );
struct md_download_context* md_download_set_iterator_get_context( const md_download_set_iterator& itr );

/// Read lock downloading contexts
int md_downloader_downloading_rlock( struct md_downloader* dl ) {
   return pthread_rwlock_rdlock( &dl->downloading_lock );
}

/// Write lock downloading contexts
int md_downloader_downloading_wlock( struct md_downloader* dl ) {
   return pthread_rwlock_wrlock( &dl->downloading_lock );
}

/// Unlock downloading contexts
int md_downloader_downloading_unlock( struct md_downloader* dl ) {
   return pthread_rwlock_unlock( &dl->downloading_lock );
}

/// Read lock the pending contexts
int md_downloader_pending_rlock( struct md_downloader* dl ) {
   return pthread_rwlock_rdlock( &dl->pending_lock );
}

/// Write lock the pending contexts
int md_downloader_pending_wlock( struct md_downloader* dl ) {
   return pthread_rwlock_wrlock( &dl->pending_lock );
}

/// Unlock the pending contexts
int md_downloader_pending_unlock( struct md_downloader* dl ) {
   return pthread_rwlock_unlock( &dl->pending_lock );
}

/// Read lock the pending contexts
int md_downloader_cancelling_rlock( struct md_downloader* dl ) {
   return pthread_rwlock_rdlock( &dl->cancelling_lock );
}

/// Write lock the cancelling context
int md_downloader_cancelling_wlock( struct md_downloader* dl ) {
   return pthread_rwlock_wrlock( &dl->cancelling_lock );
}

/// Unlock the cancelling context
int md_downloader_cancelling_unlock( struct md_downloader* dl ) {
   return pthread_rwlock_unlock( &dl->cancelling_lock );
}

/// Allocate a downloader
struct md_downloader* md_downloader_new() {
   return SG_CALLOC( struct md_downloader, 1 );
}

/**
 * @brief Set up a downloader
 * @retval 0 Success
 * @retval -ENOMEM Out of Memory
 */
int md_downloader_init( struct md_downloader* dl, char const* name ) {

   int rc = 0;

   memset( dl, 0, sizeof(struct md_downloader) );

   rc = pthread_rwlock_init( &dl->downloading_lock, NULL );
   if( rc != 0 ) {

      return -rc;
   }

   rc = pthread_rwlock_init( &dl->pending_lock, NULL );
   if( rc != 0 ) {

      pthread_rwlock_destroy( &dl->downloading_lock );
      return -rc;
   }

   rc = pthread_rwlock_init( &dl->cancelling_lock, NULL );
   if( rc != 0 ) {

      pthread_rwlock_destroy( &dl->downloading_lock );
      pthread_rwlock_destroy( &dl->pending_lock );
      return -rc;
   }

   dl->curlm = curl_multi_init();

   dl->name = SG_strdup_or_null( name );
   dl->downloading = SG_safe_new( md_downloading_map_t() );
   dl->pending = SG_safe_new( md_pending_set_t() );
   dl->cancelling = SG_safe_new( md_pending_set_t() );

   if( dl->name == NULL || dl->downloading == NULL || dl->pending == NULL || dl->cancelling == NULL || dl->curlm == NULL ) {

      SG_safe_free( dl->name );
      SG_safe_free( dl->downloading );
      SG_safe_free( dl->pending );
      SG_safe_free( dl->cancelling );

      if( dl->curlm != NULL ) {
         curl_multi_cleanup( dl->curlm );
      }

      pthread_rwlock_destroy( &dl->downloading_lock );
      pthread_rwlock_destroy( &dl->pending_lock );
      pthread_rwlock_destroy( &dl->cancelling_lock );

      return -ENOMEM;
   }

   dl->inited = true;
   return 0;
}

/**
 * @brief Start up a downloader
 * @retval 0 Success
 * @retval -1 if we failed to start the thread, or if we're already running
 */
int md_downloader_start( struct md_downloader* dl ) {

   int rc = 0;
   if( !dl->running ) {

      dl->running = true;

      rc = md_start_thread( &dl->thread, md_downloader_main, dl, false );
      if( rc < 0 ) {

         SG_error("%s: md_start_thread rc = %d\n", dl->name, rc);
         dl->running = false;
         return -1;
      }
   }

   return 0;
}

/**
 * @brief Stop a downloader
 * @retval 0 Success
 * @retval <0 Failed to join with the downloader thread
 */
int md_downloader_stop( struct md_downloader* dl ) {

   if( dl->running ) {
      dl->running = false;

      int rc = pthread_join( dl->thread, NULL );
      if( rc != 0 ) {
         SG_error("%s: pthread_join rc = %d\n", dl->name, rc );
         return rc;
      }
   }

   return 0;
}

/**
 * @brief Signal every element of a pending_set by incrementing its semaphore
 * @note Always succeeds
 */
static int md_signal_pending_set( md_pending_set_t* ps ) {

   // signal each waiting thread
   for( md_pending_set_t::iterator itr = ps->begin(); itr != ps->end(); itr++ ) {

      struct md_download_context* dlctx = *itr;

      if( dlctx != NULL ) {
         SG_debug("Wakeup %p\n", dlctx);
         sem_post( &dlctx->sem );
      }
   }

   return 0;
}

/**
 * @brief Shut down a downloader
 *
 * Stops all CURL transfers abruptly.
 * @note The downloader must be stopped first.
 * @retval 0 Success
 * @retval -EINVAL The downloader was running, or if the downloader was never initialized
 */
int md_downloader_shutdown( struct md_downloader* dl ) {

   if( dl->running ) {
      // not stopped yet
      return -EINVAL;
   }

   if( !dl->inited ) {
      // not initialized
      return -EINVAL;
   }

   // destroy downloading
   md_downloader_downloading_wlock( dl );

   dl->inited = false;

   if( dl->downloading != NULL ) {

      // remove each running download and signal the waiting threads
      for( md_downloading_map_t::iterator itr = dl->downloading->begin(); itr != dl->downloading->end(); itr++ ) {

         struct md_download_context* dlctx = itr->second;

         curl_multi_remove_handle( dl->curlm, dlctx->curl );

         sem_post( &dlctx->sem );
      }

      dl->downloading->clear();

      delete dl->downloading;
      dl->downloading = NULL;
   }

   if( dl->curlm != NULL ) {
      curl_multi_cleanup( dl->curlm );
      dl->curlm = NULL;
   }

   md_downloader_downloading_unlock( dl );

   // destroy pending
   md_downloader_pending_wlock( dl );

   if( dl->pending != NULL ) {

      // signal each waiting thread
      md_signal_pending_set( dl->pending );

      dl->pending->clear();
      delete dl->pending;
      dl->pending = NULL;
   }

   md_downloader_pending_unlock( dl );

   md_downloader_cancelling_wlock( dl );

   // destroy cancelling
   if( dl->cancelling != NULL ) {

      // signal each waiting thread
      md_signal_pending_set( dl->cancelling );

      dl->cancelling->clear();
      delete dl->cancelling;
      dl->cancelling = NULL;
   }

   md_downloader_cancelling_unlock( dl );

   // misc
   if( dl->name ) {
      free( dl->name );
      dl->name = NULL;
   }

   pthread_rwlock_destroy( &dl->downloading_lock );
   pthread_rwlock_destroy( &dl->pending_lock );
   pthread_rwlock_destroy( &dl->cancelling_lock );

   memset( dl, 0, sizeof(struct md_downloader) );

   return 0;
}

/**
 * @brief Get run state of downloader
 * @retval True Running
 * @retval False Not running
 */
bool md_downloader_is_running( struct md_downloader* dl ) {
   return dl->running;
}

/**
 * @brief Insert a pending context.
 *
 * This increments the download context's reference count.
 * @retval 0 Success, and mark it as pending and unsafe to free Success
 * @retval -EPERM The downloader is not running
 * @retval -EINVAL The download context is finalized, already pending, or already cancelling
 * @retval -ENOMEM Out of Memory
 */
int md_downloader_insert_pending( struct md_downloader* dl, struct md_download_context* dlctx ) {

   int rc = 0;

   md_downloader_pending_wlock( dl );

   if( !dl->running ) {
      md_downloader_pending_unlock( dl );
      return -EPERM;
   }

   pthread_mutex_lock( &dlctx->finalize_lock );

   if( dlctx->finalized ) {
      md_downloader_pending_unlock( dl );

      pthread_mutex_unlock( &dlctx->finalize_lock );
      return -EINVAL;
   }

   if( dlctx->pending || dlctx->cancelling ) {
      md_downloader_pending_unlock( dl );

      pthread_mutex_unlock( &dlctx->finalize_lock );
      return -EINVAL;
   }

   dlctx->pending = true;

   try {
      dl->pending->insert( dlctx );
   }
   catch( bad_alloc& ba ) {

      dlctx->pending = false;
      rc = -ENOMEM;
   }

   // reference--the downloader has a ref to it.
   dlctx->ref_count++;
   SG_debug("download %p ref %d\n", dlctx, dlctx->ref_count );

   pthread_mutex_unlock( &dlctx->finalize_lock );

   md_downloader_pending_unlock( dl );

   if( rc == 0 ) {
      dl->has_pending = true;

      SG_debug("Start download context %p\n", dlctx );
   }

   return rc;
}

/**
 * @brief Insert a context to cancel
 * @retval 0 Success, and mark the download context as cancelling
 * @retval -EPERM The downloader was not running
 * @retval -EINPROGRESS The download context is already cancelling
 * @retval -ENOMEM Out of Memory
 */
int md_downloader_insert_cancelling( struct md_downloader* dl, struct md_download_context* dlctx ) {

   int rc = 0;
   SG_debug("Cancel download context %p\n", dlctx );

   md_downloader_cancelling_wlock( dl );

   if( !dl->running ) {
      md_downloader_cancelling_unlock( dl );
      return -EPERM;
   }

   pthread_mutex_lock( &dlctx->finalize_lock );

   if( dlctx->finalized ) {
      md_downloader_cancelling_unlock( dl );
      pthread_mutex_unlock( &dlctx->finalize_lock );

      SG_warn("Download context %p is already finalized\n", dlctx );
      return 0;
   }

   if( dlctx->cancelling ) {
      md_downloader_cancelling_unlock( dl );
      pthread_mutex_unlock( &dlctx->finalize_lock );

      SG_warn("Download context %p is already cancelling\n", dlctx );
      return -EINPROGRESS;
   }

   dlctx->cancelling = true;

   try {
      if( !dlctx->pending ) {
         dl->cancelling->insert( dlctx );
      }
   }
   catch( bad_alloc& ba ) {
      rc = -ENOMEM;
      dlctx->cancelling = false;
   }

   // reference this--the downloader has a ref to it
   dlctx->ref_count++;
   SG_debug("download %p ref %d\n", dlctx, dlctx->ref_count );

   pthread_mutex_unlock( &dlctx->finalize_lock );

   dl->has_cancelling = true;

   md_downloader_cancelling_unlock( dl );

   return rc;
}

/**
 * @brief Add all pending to downloading
 * @note The downloader must be write-locked for downloading
 * @retval 0 Success
 * @retval -EPERM Failed to insert the curl handle
 * @retval -ENOMEM Out of Memory
 */
int md_downloader_start_all_pending( struct md_downloader* dl ) {

   int rc = 0;

   if( dl->has_pending ) {

      md_downloader_pending_wlock( dl );

      for( md_pending_set_t::iterator itr = dl->pending->begin(); itr != dl->pending->end(); itr++ ) {

         struct md_download_context* dlctx = *itr;

         if( dlctx == NULL ) {
            continue;
         }

         pthread_mutex_lock( &dlctx->finalize_lock );

         if( md_download_context_finalized( dlctx ) ) {

            pthread_mutex_unlock( &dlctx->finalize_lock );
            continue;
         }

         if( dlctx->cancelling ) {

            // got cancelled quickly after insertion
            dlctx->cancelled = true;
            dlctx->cancelling = false;

            pthread_mutex_unlock( &dlctx->finalize_lock );

            rc = md_downloader_finalize_download_context( dlctx, -EAGAIN );
            if( rc > 0 ) {

               // this was the last reference to the download context
               CURL* curl = NULL;
               md_download_context_free( dlctx, &curl );
               curl_easy_cleanup( curl );
               SG_safe_free( dlctx );       // allowed, since dlctx can only be heap-allocated

               rc = 0;
            }

            continue;
         }

         // add the handle
         rc = curl_multi_add_handle( dl->curlm, dlctx->curl );
         if( rc != 0 ) {

            SG_error("curl_multi_add_handle( %p ) rc = %d\n", dlctx, rc );

            rc = -EPERM;
            pthread_mutex_unlock( &dlctx->finalize_lock );
            break;
         }

         dlctx->running = true;
         dlctx->pending = false;

         pthread_mutex_unlock( &dlctx->finalize_lock );

         try {
            (*dl->downloading)[ dlctx->curl ] = dlctx;
         }
         catch( bad_alloc& ba ) {
            rc = -ENOMEM;
            break;
         }
      }

      dl->pending->clear();

      dl->has_pending = false;

      md_downloader_pending_unlock( dl );
   }

   return rc;
}

/**
 * @brief Remove all cancelling downloads from downloading
 *
 * Unreferences downloads and frees them if needed.
 * @note The downloader must be write-locked for downloading
 * @retval 0 Success
 * @retval -EPERM Failed to remove the curl handle
 * @retval -ENOMEM Out of Memory
 */
int md_downloader_end_all_cancelling( struct md_downloader* dl ) {

   int rc = 0;

   if( dl->has_cancelling ) {

      md_downloader_cancelling_wlock( dl );

      for( md_pending_set_t::iterator itr = dl->cancelling->begin(); itr != dl->cancelling->end(); itr++ ) {

         struct md_download_context* dlctx = *itr;

         if( dlctx == NULL ) {
            continue;
         }

         pthread_mutex_lock( &dlctx->finalize_lock );

         rc = curl_multi_remove_handle( dl->curlm, dlctx->curl );
         if( rc != 0 ) {

            SG_error("curl_multi_remove_handle( %p ) rc = %d\n", dlctx, rc );

            rc = -EPERM;
            pthread_mutex_unlock( &dlctx->finalize_lock );
            continue;
         }

         // NOTE: this is the unref that the client would have called, had they not cancelled
         dlctx->ref_count--;
         SG_debug("download %p ref %d\n", dlctx, dlctx->ref_count );
         dl->downloading->erase( dlctx->curl );

         // update state
         dlctx->cancelled = true;
         dlctx->cancelling = false;

         // struct md_download_set* dlset = dlctx->dlset;

         pthread_mutex_unlock( &dlctx->finalize_lock );

         // finalize, with -EAGAIN
         rc = md_downloader_finalize_download_context( dlctx, -EAGAIN );
         if( rc > 0 ) {

            // this was the last reference to the download context
            CURL* curl = NULL;
            md_download_context_free( dlctx, &curl );
            curl_easy_cleanup( curl );
            SG_safe_free( dlctx );      // allowed, since dlctx can only be heap-allocated

            rc = 0;
         }
         /*
         // wake up the set waiting on this dlctx
         if( dlset != NULL ) {

            rc = md_download_set_wakeup( dlset );
            if( rc != 0 ) {

               SG_error("md_download_set_wakeup( %p ) rc = %d\n", dlset, rc );
               rc = 0;
               continue;
            }
         }
         */
      }

      dl->cancelling->clear();

      dl->has_cancelling = false;

      md_downloader_cancelling_unlock( dl );
   }

   return 0;
}

/**
 * @brief Download data to a response buffer
 * @retval Size Success
 * @retval 0 Out of Memory
 */
size_t md_get_callback_response_buffer( void* stream, size_t size, size_t count, void* user_data ) {

   md_response_buffer_t* rb = (md_response_buffer_t*)user_data;

   size_t realsize = size * count;
   char* buf = SG_CALLOC( char, realsize );

   if( buf == NULL ) {
      // OOM
      return 0;
   }

   memcpy( buf, stream, realsize );

   try {
      rb->push_back( md_buffer_segment_t( buf, realsize ) );
   }
   catch( bad_alloc& ba ) {

      SG_safe_free( buf );
      return 0;
   }

   return realsize;
}

/**
 * @brief Download to a bound response buffer
 * @retval Size Success
 * @retval 0 Out of Memory
 */
size_t md_get_callback_bound_response_buffer( void* stream, size_t size, size_t count, void* user_data ) {

   struct md_bound_response_buffer* brb = (struct md_bound_response_buffer*)user_data;

   // SG_debug("size = %zu, count = %zu, max_size = %ld, size = %ld\n", size, count, brb->max_size, brb->size );

   if( brb->size >= brb->max_size ) {
      return 0;
   }

   off_t realsize = size * count;
   if( brb->max_size >= 0 && (off_t)(brb->size + realsize) > brb->max_size ) {

      realsize = brb->max_size - brb->size;
      if( realsize < 0 ) {
         return 0;
      }
   }

   char* buf = SG_CALLOC( char, realsize );
   if( buf == NULL ) {
      return 0;
   }

   memcpy( buf, stream, realsize );

   try {
      brb->rb->push_back( md_buffer_segment_t( buf, realsize ) );
   }
   catch( bad_alloc& ba ) {

      SG_safe_free( buf );
      return 0;
   }

   brb->size += realsize;

   return realsize;
}

/**
 * @brief Alloc a download context
 */
struct md_download_context* md_download_context_new() {
   return SG_CALLOC( struct md_download_context, 1 );
}

/**
 * @brief Initialize a download context.  Takes a CURL handle from the client.
 *
   The only things it sets in the CURL handle are:
   CURLOPT_WRITEDATA
   CURLOPT_WRITEFUNCTION
 * @retval 0 Success
 * @retval -ENOMEM Out of Memory
 * @retval -errno on failure to initialize a pthread lock
 */
int md_download_context_init( struct md_download_context* dlctx, CURL* curl, off_t max_len, void* cls ) {

   SG_debug("Initialize download context %p\n", dlctx );

   int rc = 0;
   memset( dlctx, 0, sizeof(struct md_download_context) );

   rc = pthread_mutex_init( &dlctx->finalize_lock, NULL );
   if( rc != 0 ) {
      return -rc;
   }

   rc = md_bound_response_buffer_init( &dlctx->brb, max_len );
   if( rc != 0 ) {

      pthread_mutex_destroy( &dlctx->finalize_lock );
      return rc;
   }

   dlctx->curl = curl;
   dlctx->cls = cls;

   sem_init( &dlctx->sem, 0, 0 );

   curl_easy_setopt( dlctx->curl, CURLOPT_WRITEDATA, (void*)&dlctx->brb );
   curl_easy_setopt( dlctx->curl, CURLOPT_WRITEFUNCTION, md_get_callback_bound_response_buffer );

   dlctx->dlset = NULL;
   dlctx->initialized = true;

   return 0;
}

/**
 * @brief Reset a download context.
 * @note Don't call this until it's finalized
 * @retval 0 Success
 * @retval -EAGAIN The download context is not finalized
 */
int md_download_context_reset( struct md_download_context* dlctx, CURL** old_curl ) {

   SG_debug("Reset download context %p\n", dlctx );

   pthread_mutex_lock( &dlctx->finalize_lock );

   if( !md_download_context_finalized( dlctx ) ) {
      SG_error("Download %p not yet finalized\n", dlctx );

      pthread_mutex_unlock( &dlctx->finalize_lock );
      return -EAGAIN;
   }

   md_response_buffer_free( dlctx->brb.rb );
   dlctx->brb.size = 0;

   curl_easy_setopt( dlctx->curl, CURLOPT_WRITEDATA, (void*)&dlctx->brb );
   curl_easy_setopt( dlctx->curl, CURLOPT_WRITEFUNCTION, md_get_callback_bound_response_buffer );

   dlctx->curl_rc = 0;
   dlctx->http_status = 0;
   dlctx->transfer_errno = 0;
   dlctx->cancelled = false;
   dlctx->finalized = false;
   dlctx->pending = false;
   dlctx->cancelling = false;
   dlctx->running = false;
   dlctx->cls = NULL;
   dlctx->ref_count = 0;
   SG_debug("download %p ref-set %d\n", dlctx, dlctx->ref_count );

   if( dlctx->effective_url != NULL ) {

      SG_safe_free( dlctx->effective_url );
   }

   if( old_curl != NULL ) {
      *old_curl = dlctx->curl;
   }

   pthread_mutex_unlock( &dlctx->finalize_lock );

   return 0;
}

/**
 * @brief Free a download context.
 *
 * If this download context was finalized, then it's guaranteed to be freed
 * @note Does *not* check reference count
 * @note Always succeeds
 * @return 0
 */
int md_download_context_free2( struct md_download_context* dlctx, CURL** curl, char const* file, int lineno ) {

   pthread_mutex_lock( &dlctx->finalize_lock );

   SG_debug("Free download context %p, from %s:%d (refcount %d)\n", dlctx, file, lineno, dlctx->ref_count );
   if( dlctx->ref_count > 0 ) {

      SG_error("BUG: download context %p has %d references\n", dlctx, dlctx->ref_count );
      exit(1);
   }

   md_bound_response_buffer_free( &dlctx->brb );

   if( curl != NULL ) {
      *curl = dlctx->curl;
   }

   if( dlctx->effective_url != NULL ) {

      SG_safe_free( dlctx->effective_url );
   }

   dlctx->curl = NULL;

   sem_destroy( &dlctx->sem );

   pthread_mutex_unlock( &dlctx->finalize_lock );
   pthread_mutex_destroy( &dlctx->finalize_lock );

   memset( dlctx, 0, sizeof(struct md_download_context));

   return 0;
}

/**
 * @brief Reference this download context
 * @note Always succeeds
 * @return 0
 */
int md_download_context_ref2( struct md_download_context* dlctx, char const* file, int lineno ) {

   pthread_mutex_lock( &dlctx->finalize_lock );

   dlctx->ref_count++;
   SG_debug("download %p ref %d (from %s:%d)\n", dlctx, dlctx->ref_count, file, lineno );

   pthread_mutex_unlock( &dlctx->finalize_lock );
   return 0;
}

/**
 * @brief Unreference this download context
 *
 * If the reference count reaches 0, then free it
 * @retval 0 Success
 * @retval 1 Success, in which case the caller should follow up this call with a call to md_download_context_free.
 */
int md_download_context_unref2( struct md_download_context* dlctx, char const* file, int lineno ) {

   pthread_mutex_lock( &dlctx->finalize_lock );

   dlctx->ref_count--;
   SG_debug("download %p ref %d (from %s:%d)\n", dlctx, dlctx->ref_count, file, lineno );

   if( dlctx->ref_count <= 0 ) {

      dlctx->ref_count = 0;

      pthread_mutex_unlock( &dlctx->finalize_lock );

      return 1;
   }

   pthread_mutex_unlock( &dlctx->finalize_lock );
   return 0;
}

/**
 * @brief Unref, and possibly free a download context if it's fully unref'ed
 * @retval 0 Success
 * @retval 1 Success and free
 */
int md_download_context_unref_free( struct md_download_context* dlctx, CURL** ret_curl ) {

   int rc = md_download_context_unref( dlctx );
   if( rc > 0 ) {

      md_download_context_free( dlctx, ret_curl );
   }

   return rc;
}

/**
 * @brief Remove a download context if it is part of a download set
 * @note Always succeeds
 * @return 0
 */
int md_download_context_clear_set( struct md_download_context* dlctx ) {

   if( dlctx != NULL && dlctx->dlset != NULL ) {
      md_download_set_clear( dlctx->dlset, dlctx );
      dlctx->dlset = NULL;
   }

   return 0;
}

/**
 * @brief Wait for a download semaphore to be signaled, up to a given number of milliseconds.
 *
 * If timeout_ms is less than or equal to 0, then wait indefinitely.
 * If timeout_ms <= 0, then use sem_wait.  Otherwise, use sem_trywait +timeout_ms seconds into the future.
 * @note This method masks -EINTR.
 * @retval 0 Success
 * @retval -ETIMEDOUT Timed out waiting
 * @retval -errno sem_timedwait(2) or sem_wait(2) failed for some other reason
 */
int md_download_sem_wait( sem_t* sem, int64_t timeout_ms ) {

   int rc = 0;

   // do we timeout the wait?
   if( timeout_ms > 0 ) {

      struct timespec abs_ts;
      clock_gettime( CLOCK_REALTIME, &abs_ts );
      abs_ts.tv_sec += timeout_ms / 1000L;
      abs_ts.tv_nsec += timeout_ms / 1000000L;

      if( abs_ts.tv_nsec >= 1000000000L) {

         // wrap around
         abs_ts.tv_nsec %= 1000000000L;
         abs_ts.tv_sec ++;
      }

      while( true ) {

         rc = sem_timedwait( sem, &abs_ts );
         if( rc == 0 ) {

            break;
         }
         else if( errno != EINTR ) {

            rc = -errno;
            SG_error("sem_timedwait errno = %d\n", rc );
            break;
         }

         // otherwise, try again if interrupted
      }
   }
   else {
      while( true ) {

         rc = sem_wait( sem );
         if( rc == 0 ) {

            break;
         }
         else if( errno != EINTR ) {

            rc = -errno;
            SG_error("sem_wait errno = %d\n", rc );
            break;
         }

         // otherwise, try again if interrupted
      }
   }

   return rc;
}

/**
 * @brief Wait for a download to finish, either in error or not
 * @return The result of waiting, NOT the result of the download (see md_download_sem_wait for possible return codes)
 * @retval 0 Success
 * @retval -ETIMEDOUT timeout_ms is >= 0 and the deadline was exceeded
 */
int md_download_context_wait( struct md_download_context* dlctx, int64_t timeout_ms ) {

   SG_debug("Wait on download context %p (%" PRIu64 " millis)\n", dlctx, timeout_ms );

   int rc = md_download_sem_wait( &dlctx->sem, timeout_ms );

   if( rc != 0 ) {
      SG_error("md_download_sem_wait rc = %d\n", rc );
   }
   return rc;
}

/**
 * @brief Wait for a download to finish within a download set, either in error or not.
 * @return The result of waiting, NOT the result of the download (see md_downlaod_sem_wait for possible return codes)
 * @retval 0 Success
 * @retval -EINVAL The download set does not have any waiting contexts
 */
int md_download_context_wait_any( struct md_download_set* dlset, int64_t timeout_ms ) {

   if( dlset->waiting == NULL ) {
      return -EINVAL;
   }

   if( dlset->waiting->size() == 0 ) {
      return 0;
   }

   SG_debug("Wait on download set %p (%zu contexts)\n", dlset, dlset->waiting->size() );

   int rc = 0;

   // wait for at least one of them to finish
   rc = md_download_sem_wait( &dlset->sem, timeout_ms );

   if( rc != 0 ) {
      SG_error("md_download_sem_wait rc = %d\n", rc );
   }

   return rc;
}

/**
 * @brief Set up a download set
 * @retval 0 Success
 * @retval -ENOMEM Out of Memory
 */
int md_download_set_init( struct md_download_set* dlset ) {

   SG_debug("Initialize download set %p\n", dlset );

   dlset->waiting = SG_safe_new( md_pending_set_t() );
   if( dlset->waiting == NULL ) {
      return -ENOMEM;
   }

   sem_init( &dlset->sem, 0, 0 );

   return 0;
}

/**
 * @brief Free a download set
 * @note Not thread safe
 * @retval 0 Success
 */
int md_download_set_free( struct md_download_set* dlset ) {

   SG_debug("Free download set %p\n", dlset );

   if( dlset->waiting ) {

      SG_safe_delete( dlset->waiting );
   }

   sem_destroy( &dlset->sem );

   memset( dlset, 0, sizeof( struct md_download_set ) );

   return 0;
}

/**
 * @brief Add a download context to a download set.
 *
 * @note Do this before starting the download, does not affect the download's reference count
 * @note Not thread safe
 * @retval 0 Success
 * @retval -ENOMEM Out of Memory
 */
int md_download_set_add( struct md_download_set* dlset, struct md_download_context* dlctx ) {

   int rc = 0;

   try {

      md_pending_set_t::iterator itr = dlset->waiting->find( dlctx );
      if( itr == dlset->waiting->end() ) {

         dlset->waiting->insert( dlctx );

         dlctx->dlset = dlset;

         SG_debug("Add download context %p to download set %p\n", dlctx, dlset );
      }
   }
   catch( bad_alloc& ba ) {
      rc = -ENOMEM;
   }

   return rc;
}

/**
 * @brief Remove a download context from a download set by value
 * @note Don't do this in e.g. a for() loop where you're iterating over download contexts
 * @note Not thread safe
 * @retval 0 Success
 * @retval -ENOMEM Out of Memory
 */
int md_download_set_clear( struct md_download_set* dlset, struct md_download_context* dlctx ) {

   try {
      if( dlctx != NULL && dlset->waiting != NULL ) {
         dlset->waiting->erase( dlctx );
      }
   }
   catch( bad_alloc& ba ) {
      return -ENOMEM;
   }

   if( dlctx != NULL ) {
      dlctx->dlset = NULL;
   }

   return 0;
}

/**
 * @brief Get number of items in a download set
 * @note Not thread safe
 * @return The number of items
 */
size_t md_download_set_size( struct md_download_set* dlset ) {

   return dlset->waiting->size();
}

/**
 * @brief Get whether this download set contain the provided download
 * @note Not thread safe
 */
bool md_download_set_contains( struct md_download_set* dlset, struct md_download_context* dlctx ) {
   return dlset->waiting->count( dlctx ) > 0;
}

/**
 * @brief Start the iteration to the provided download set
 * @return An iterator to the waiting set
 */
md_download_set_iterator md_download_set_begin( struct md_download_set* dlset ) {
   return dlset->waiting->begin();
}

/**
 * @brief Stop the iteration to the provided download set
 * @return An interator to the waiting set
 */
md_download_set_iterator md_download_set_end( struct md_download_set* dlset ) {
   return dlset->waiting->end();
}

/**
 * @brief Dereference the provided download set iterator
 * @return A pointer to the download context
 */
struct md_download_context* md_download_set_iterator_get_context( const md_download_set_iterator& itr ) {
   return *itr;
}

/**
 * @brief Begin downloading something
 *
 * This will ref the download (i.e. use it in conjunction with md_download_loop_watch)
 * @retval 0 Success
 * @retval !0 Failed to insert the download into the downloader
 */
int md_download_context_start( struct md_downloader* dl, struct md_download_context* dlctx ) {

   md_download_context_ref( dlctx );

   // enqueue the context into the downloader
   int rc = md_downloader_insert_pending( dl, dlctx );
   if( rc != 0 ) {

      SG_error("%s: md_downloader_insert_pending( %p ) rc = %d\n", dl->name, dlctx, rc );
      return rc;
   }

   return 0;
}

/**
 * @brief Cancel download
 * @retval 0 Success, or the download is already cancelled or finalized
 * @retval -EPERM The downloader was not running
 * @retval <0 Failed to insert (but mask EINPROGRESS) and/or wait for the download to get cancelled
 */
int md_download_context_cancel( struct md_downloader* dl, struct md_download_context* dlctx ) {

   if( !dl->running ) {
      return -EPERM;
   }

   pthread_mutex_lock( &dlctx->finalize_lock );

   if( dlctx->cancelled || dlctx->finalized || dlctx->cancelling ) {

      pthread_mutex_unlock( &dlctx->finalize_lock );

      SG_debug("already cancelled %p\n", dlctx );
      return 0;
   }

   struct md_download_set* dlset = dlctx->dlset;

   pthread_mutex_unlock( &dlctx->finalize_lock );

   int rc = md_downloader_insert_cancelling( dl, dlctx );

   if( rc != 0 && rc != -EINPROGRESS ) {

      // should only happen if the downloader isn't running
      SG_error("md_downloader_insert_cancelling(%p) rc = %d\n", dlctx, rc );
      return rc;
   }
   else {

      // EINPROGRESS is okay
      rc = 0;
   }

   SG_debug("Wait for cancelling download %p (set %p)\n", dlctx, dlset );
   rc = md_download_context_wait( dlctx, -1 );
   if( rc != 0 ) {

      // should only happen if the semaphore is invalid
      SG_error("md_download_context_wait(%p) rc = %d\n", dlctx, rc );
   }
   else {
      SG_debug("cancelled %p\n", dlctx );
   }

   return rc;
}

/**
 * @brief Release a waiting context set, given one of it's now-finished entries.
 * @retval 0 Success
 * @retval -EINVAL The download set is NULL
 */
int md_download_set_wakeup( struct md_download_set* dlset ) {

   SG_debug("Wake up download set %p\n", dlset );

   int rc = 0;

   if( dlset == NULL ) {
      return -EINVAL;
   }

   sem_post( &dlset->sem );

   return rc;
}

/**
 * @brief Run multiple downloads
 *
 * @note Download must be write-locked for downloading
 * @retval >0 Curl failed
 * @retval <0 select() failed
 */
int md_downloader_run_multi( struct md_downloader* dl ) {

   int still_running = 0;
   int rc = 0;
   struct timeval timeout;

   fd_set fdread;
   fd_set fdwrite;
   fd_set fdexcep;

   FD_ZERO( &fdread );
   FD_ZERO( &fdwrite );
   FD_ZERO( &fdexcep );

   int maxfd = -1;

   long curl_timeo = -1;

   // download for a bit
   rc = curl_multi_perform( dl->curlm, &still_running );
   if( rc != 0 ) {

      SG_error("%s: curl_multi_perform rc = %d\n", dl->name, rc );
      return rc;
   }

   // don't wait more than 5ms
   timeout.tv_sec = 0;
   timeout.tv_usec = 5000;      // 5ms

   rc = curl_multi_timeout( dl->curlm, &curl_timeo );
   if( rc != 0 ) {

      SG_error("%s: curl_multi_timeout rc = %d\n", dl->name, rc );
      return rc;
   }

   if( curl_timeo > 0 ) {
      timeout.tv_sec = curl_timeo / 1000;
      if( timeout.tv_sec > 0 ) {
         timeout.tv_sec = 0;
      }

      // no more than 5ms
      timeout.tv_usec = MIN( (curl_timeo % 1000) * 1000, 5000 );
   }

   // get fd set
   rc = curl_multi_fdset( dl->curlm, &fdread, &fdwrite, &fdexcep, &maxfd );
   if( rc != 0 ) {

      SG_error("%s: curl_multi_fdset rc = %d\n", dl->name, rc );
      return rc;
   }

   // select on them
   rc = select( maxfd + 1, &fdread, &fdwrite, &fdexcep, &timeout );
   if( rc < 0 ) {

      rc = -errno;
      SG_error("%s: select rc = %d\n", dl->name, rc );
      return rc;
   }
   else {
      rc = 0;
   }
   return rc;
}

/**
 * @brief Finalize a download context.
 *
 * On success, populate it with the curl rc, the HTTP status, the transfer OS errno, and the effective URL
 * http_status == -1 means that we couldn't get the HTTP status
 * os_errno == EIO means that we couldn't get the OS errno or the URL
 * @retval 0 Success, or if the download was already finalized (if the latter, then the fields in dlctx are untouched)
 * @retval 1 Success, in which case the caller should free the download context
 * @retval -ENOMEM Out of Memory
 */
int md_downloader_finalize_download_context( struct md_download_context* dlctx, int curl_rc ) {

   pthread_mutex_lock( &dlctx->finalize_lock );

   // sanity check
   if( md_download_context_finalized( dlctx ) ) {

      SG_debug("Download context %p already finalized\n", dlctx );
      pthread_mutex_unlock( &dlctx->finalize_lock );
      return 0;
   }

   int rc = 0;

   // check HTTP code
   long http_status = 0;
   long os_errno = 0;
   char* url = NULL;
   char* url_dup = NULL;

   rc = curl_easy_getinfo( dlctx->curl, CURLINFO_RESPONSE_CODE, &http_status );
   if( rc != 0 ) {

      SG_error("curl_easy_getinfo(%p) rc = %d\n", dlctx, rc );
      http_status = -1;
   }

   // check error code
   if( rc != 0 ) {
      rc = curl_easy_getinfo( dlctx->curl, CURLINFO_OS_ERRNO, &os_errno );

      if( rc != 0 ) {
         SG_error("curl_easy_getinfo(%p) rc = %d\n", dlctx, rc );
         os_errno = EIO;
      }
   }

   // get URL
   rc = curl_easy_getinfo( dlctx->curl, CURLINFO_EFFECTIVE_URL, &url );

   if( rc != 0 || url == NULL ) {
      SG_error("curl_easy_getinfo(%p) rc = %d\n", dlctx, rc );
      os_errno = EIO;
   }

   if( url != NULL ) {
      url_dup = SG_strdup_or_null( url );
      if( url_dup == NULL ) {

         os_errno = -ENOMEM;
      }
   }

   dlctx->curl_rc = curl_rc;
   dlctx->http_status = (int)http_status;
   dlctx->transfer_errno = (int)os_errno;
   dlctx->effective_url = NULL;

   if( url_dup != NULL ) {
      dlctx->effective_url = url_dup;

      SG_debug("Finalized download context %p (%s)\n", dlctx, dlctx->effective_url );
   }
   else {
      SG_debug("Finalized download context %p\n", dlctx );
   }

   dlctx->finalized = true;
   dlctx->running = false;

   // unreferenced
   dlctx->ref_count--;
   SG_debug("download %p ref %d\n", dlctx, dlctx->ref_count );

   if( dlctx->ref_count <= 0 ) {

      // caller should free
      rc = 1;
   }

   // wake up threads waiting for this download set
   if( dlctx->dlset != NULL ) {

      rc = md_download_set_wakeup( dlctx->dlset );
      if( rc != 0 ) {

         SG_error("md_download_set_wakeup( %p ) rc = %d\n", dlctx->dlset, rc );
         rc = 0;
      }
   }

   // wake up threads waitin for this specific download
   sem_post( &dlctx->sem );

   pthread_mutex_unlock( &dlctx->finalize_lock );

   return rc;
}

/**
 * @brief Finalize all finished downloads
 *
 * Try to remove all downloads, even if we fail to remove some.
 * @note Download must be write-locked for downloads
 * @retval 0 Success
 * @retval -ENOMEM Out of Memory
 * @retval positive if curl did something fatal
 */
int md_downloader_finalize_download_contexts( struct md_downloader* dl ) {
   CURLMsg* msg = NULL;
   int msgs_left = 0;
   int rc = 0;

   do {
      msg = curl_multi_info_read( dl->curlm, &msgs_left );

      if( msg == NULL ) {
         // no messages
         break;
      }

      if( msg->msg == CURLMSG_DONE ) {
         // a transfer finished.  Find out which one
         md_downloading_map_t::iterator itr = dl->downloading->find( msg->easy_handle );
         if( itr != dl->downloading->end() ) {

            // found!
            struct md_download_context* dlctx = itr->second;

            // get this now, before removing it from the curlm handle
            int result = msg->data.result;

            try {
               // remove from the downloader
               dl->downloading->erase( itr );
            }
            catch( bad_alloc& ba ) {
               rc = -ENOMEM;
               break;
            }

            if( dlctx == NULL ) {

               SG_warn("no download context for curl handle %p\n", msg->easy_handle);

               rc = curl_multi_remove_handle( dl->curlm, msg->easy_handle );
               if( rc != 0 ) {

                  SG_error("curl_multi_remove_handle( %p ) rc = %d\n", msg->easy_handle, rc );
                  rc = 0;
               }

               continue;
            }

            if( dlctx->curl == NULL ) {
               SG_error("BUG: curl handle of download context %p is NULL\n", dlctx );

               rc = curl_multi_remove_handle( dl->curlm, msg->easy_handle );
            }
            else {
               rc = curl_multi_remove_handle( dl->curlm, dlctx->curl );
               if( rc != 0 ) {
                  SG_error("curl_multi_remove_handle(%p) rc = %d\n", msg->easy_handle, rc );
                  rc = 0;
               }
            }

            // get the download set from this dlctx, so we can awaken it later
            // struct md_download_set* dlset = dlctx->dlset;

            // finalize the download context, unref'ing it
            rc = md_downloader_finalize_download_context( dlctx, result );
            if( rc < 0 ) {

               SG_error("%s: md_downloader_finalize_download_context rc = %d\n", dl->name, rc );
               rc = 0;
               continue;
            }

            if( rc > 0 ) {

               // this was the last reference to the download context.  We should free it.
               CURL* curl = NULL;
               md_download_context_free( dlctx, &curl );
               curl_easy_cleanup( curl );
               SG_safe_free( dlctx );       // allowed, since dlctx can only be heap-allocated

               rc = 0;
            }

            /*
            // wake up the set waiting on this dlctx
            if( dlset != NULL ) {

               rc = md_download_set_wakeup( dlset );
               if( rc != 0 ) {

                  SG_error("md_download_set_wakeup( %p ) rc = %d\n", dlset, rc );
                  rc = 0;
                  continue;
               }
            }
            */
         }
      }

   } while( msg != NULL );

   return rc;
}

/**
 * @brief Main downloader loop
 * @return NULL
 */
static void* md_downloader_main( void* arg ) {

   struct md_downloader* dl = (struct md_downloader*)arg;

   SG_debug("%s: starting\n", dl->name );

   int rc = 0;

   while( dl->running ) {

      md_downloader_downloading_wlock( dl );

      // add all pending downloads to this downloader
      rc = md_downloader_start_all_pending( dl );
      if( rc != 0 ) {
         SG_error("%s: md_downloader_start_all_pending rc = %d\n", dl->name, rc );
      }

      // remove all cancelled downloads from this downloader
      rc = md_downloader_end_all_cancelling( dl );
      if( rc != 0 ) {
         SG_error("%s: md_downloader_end_all_cancelling rc = %d\n", dl->name, rc );
      }

      // download for a bit
      rc = md_downloader_run_multi( dl );
      if( rc != 0 ) {
         SG_error("%s: md_downloader_run_multi rc = %d\n", dl->name, rc );
      }

      // finalize any completed downloads
      rc = md_downloader_finalize_download_contexts( dl );
      if( rc != 0 ) {
         SG_error("%s: md_downloader_finalize_download_contexts rc = %d\n", dl->name, rc );
      }

      md_downloader_downloading_unlock( dl );

      // give the md_downloader_stop() method a chance to preempt the main method
   }

   SG_debug("%s: exiting\n", dl->name );
   return NULL;
}

/**
 * @brief Consolidate and write back the buffer
 * @retval 0 Success, and set *buf and *buf_len accordingly
 * @retval -ENOMEM Out of Memory
 */
int md_download_context_get_buffer( struct md_download_context* dlctx, char** buf, off_t* buf_len ) {

   *buf = md_response_buffer_to_string( dlctx->brb.rb );
   *buf_len = md_response_buffer_size( dlctx->brb.rb );

   if( *buf == NULL ) {
      return -ENOMEM;
   }

   return 0;
}

/**
 * @brief Get the http status
 * @return The positive HTTP status, Success
 * @retval -EAGAIN The download context was not finalized
 */
int md_download_context_get_http_status( struct md_download_context* dlctx ) {

   if( !dlctx->finalized ) {

      return -EAGAIN;
   }

   return dlctx->http_status;
}

/**
 * @brief Get the errno
 * @return The transfer error code (non-negative number) on sucess
 * @retval -EAGAIN The download context was not finalized
 */
int md_download_context_get_errno( struct md_download_context* dlctx ) {
   if( !dlctx->finalized ) {
      return -EAGAIN;
   }

   return dlctx->transfer_errno;
}

/**
 * @brief Get the curl rc
 * @retval >=0 Success
 * @retval -EAGAIN The download was not finalized
 */
int md_download_context_get_curl_rc( struct md_download_context* dlctx ) {
   if( !dlctx->finalized ) {
      return -EAGAIN;
   }

   return dlctx->curl_rc;
}

/**
 * @brief Get the effective URL
 * @retval 0 Success, and set *url to a malloc'ed copy of the URL (if there was a URL to begin with)
 * @retval -ENOMEM Out of Memory
 * @retval -EAGAIN The download was not finalized
 */
int md_download_context_get_effective_url( struct md_download_context* dlctx, char** url ) {
   if( !dlctx->finalized ) {
      return -EAGAIN;
   }

   if( dlctx->effective_url == NULL ) {
      *url = NULL;
   }
   else {

      *url = SG_strdup_or_null( dlctx->effective_url );
      if( *url == NULL ) {
         return -ENOMEM;
      }
   }

   return 0;
}

/**
 * @brief Get the download handle's curl handle
 * @return The curl handle
 */
CURL* md_download_context_get_curl( struct md_download_context* dlctx ) {
   return dlctx->curl;
}

/**
 * @brief Get the download handle's cls
 * @return Pointer to the cls
 */
void* md_download_context_get_cls( struct md_download_context* dlctx ) {
   return dlctx->cls;
}

/**
 * @brief Set the donwload handle's cls
 * @note Not thread-safe; only use it if you know what you're doing!
 * @note Always succeeds
 */
void md_download_context_set_cls( struct md_download_context* dlctx, void* new_cls ) {
   dlctx->cls = new_cls;
}

/**
 * @brief State of download context
 *
 * Check if the download context worked
 * @retval True The download's curl rc is 0, its transfer errno is 0, and its HTTP status is the desired HTTP status
 * @retval False Otherwise
 */
bool md_download_context_succeeded( struct md_download_context* dlctx, int desired_HTTP_status ) {
   return (dlctx->curl_rc == 0 && dlctx->transfer_errno == 0 && dlctx->http_status == desired_HTTP_status);
}

/**
 * @brief Check if download is finalized
 * @retval True Finalized
 * @retval False Not finalized
 */
bool md_download_context_finalized( struct md_download_context* dlctx ) {
   return dlctx->finalized;
}

/**
 * @brief Check if download is running
 * @retval True Running
 * @retval False Not running
 */
bool md_download_context_running( struct md_download_context* dlctx ) {
   return dlctx->running;
}

/**
 * @brief Check if download is pending
 * @retval True Pending
 * @retval False Not pending
 */
bool md_download_context_pending( struct md_download_context* dlctx ) {
   return dlctx->pending;
}

/**
 * @brief Check if download was cancelled
 * @retval True Cancelled
 * @retval False Not cancelled
 */
bool md_download_context_cancelled( struct md_download_context* dlctx ) {
   return dlctx->cancelled;
}

/**
 * @brief Check if a download is initialized
 * @retval True Initialized
 * @retval False Not initialized
 */
bool md_download_context_initialized( struct md_download_context* dlctx ) {
   return dlctx->initialized;
}

/**
 * @brief Run a single download context
 * @note dlctx *cannot* be locked
 * @retval 0 Successful finalization, even if the download failed
 * @retval 1 Successful finalization, even if the download failed, in which case the caller should free the download context since it's been fully unref'ed
 * @retval <0 Otherwise (see md_downloader_finalize_download_context)
 */
int md_download_context_run( struct md_download_context* dlctx ) {

   int rc = 0;

   dlctx->running = true;

   rc = curl_easy_perform( dlctx->curl );
   if( rc != 0 ) {

      SG_error("curl_easy_perform( %p ) rc = %d\n", dlctx, rc );
   }

   rc = md_downloader_finalize_download_context( dlctx, rc );

   return rc;
}

/**
 * @brief Set sockopt for curl
 * @todo Test various flags for curl sockopt, TCP_CORK or TCP_NODELAY
 * @retval CURL_SOCKOPT_OK
 */
static int md_curl_sockopt( void* userdata, curl_socket_t sockfd, curlsocktype purpose ) {

   int rc = CURL_SOCKOPT_OK;

   if( purpose == CURLSOCKTYPE_IPCXN ) {

      // opening a socket
      // TODO: experiment with different flags here--maybe TCP_CORK or TCP_NODELAY
      rc = CURL_SOCKOPT_OK;
   }

   return rc;
}

/**
 * @brief Initialze a curl handle
 */
static void md_init_curl_handle2( CURL* curl_h, char const* url, time_t query_timeout, bool ssl_verify_peer, int64_t transfer_timeout ) {

   curl_easy_setopt( curl_h, CURLOPT_NOPROGRESS, 1L );   // no progress bar
   curl_easy_setopt( curl_h, CURLOPT_USERAGENT, "Syndicate-Gateway/1.0");

   if( url != NULL ) {
      curl_easy_setopt( curl_h, CURLOPT_URL, url );
   }

   curl_easy_setopt( curl_h, CURLOPT_FOLLOWLOCATION, 1L );
   curl_easy_setopt( curl_h, CURLOPT_MAXREDIRS, 10L );
   curl_easy_setopt( curl_h, CURLOPT_NOSIGNAL, 1L );
   curl_easy_setopt( curl_h, CURLOPT_CONNECTTIMEOUT, query_timeout );
   curl_easy_setopt( curl_h, CURLOPT_FILETIME, 1L );

   if( url != NULL && strncasecmp( url, "https", 5 ) == 0 ) {
      curl_easy_setopt( curl_h, CURLOPT_USE_SSL, CURLUSESSL_ALL );
      curl_easy_setopt( curl_h, CURLOPT_SSL_VERIFYPEER, ssl_verify_peer ? 1L : 0L );
      curl_easy_setopt( curl_h, CURLOPT_SSL_VERIFYHOST, 2L );
   }
   else {
      curl_easy_setopt( curl_h, CURLOPT_USE_SSL, CURLUSESSL_NONE );
      curl_easy_setopt( curl_h, CURLOPT_SSL_VERIFYPEER, 0L );
      curl_easy_setopt( curl_h, CURLOPT_SSL_VERIFYHOST, 0L );
   }

   curl_easy_setopt( curl_h, CURLOPT_SOCKOPTFUNCTION, md_curl_sockopt );

   // stop-gap
   curl_easy_setopt( curl_h, CURLOPT_TIMEOUT, transfer_timeout );

   curl_easy_setopt( curl_h, CURLOPT_VERBOSE, 1L );
}

/**
 * @brief Initialze a curl handle
 * @note Always succeeds
 */
void md_init_curl_handle( struct md_syndicate_conf* conf, CURL* curl_h, char const* url, time_t query_timeout ) {
   md_init_curl_handle2( curl_h, url, query_timeout, conf->verify_peer, conf->transfer_timeout );
}

/**
 * @brief Interpret error messages from a download context into an appropriate return code for the downloader.
 * @retval 0 Success
 * @retval -EAGAIN The download should be retried
 * @retval -ETIMEDOUT Operation timeout
 * @retval -EREMOTEIO The HTTP status was >= 500, or an indeterminate error occurred but errno was not set.
 * @retval -http_status The HTTP status was between 400 and 499
 */
int md_download_interpret_errors( int http_status, int curl_rc, int os_err ) {

   int rc = 0;

   if( http_status == SG_HTTP_TRYAGAIN ) {

      return -EAGAIN;
   }

   if( curl_rc == CURLE_OPERATION_TIMEDOUT || os_err == -ETIMEDOUT || curl_rc == CURLE_GOT_NOTHING ) {

      return -ETIMEDOUT;
   }

   // serious error?
   if( http_status >= 500 ) {

      return -EREMOTEIO;
   }

   // some other error?
   if( http_status != 200 || curl_rc != 0 ) {

      if( http_status >= 400 && http_status <= 499 ) {
         rc = -http_status;
      }
      else if( os_err != 0 ) {
         rc = -os_err;
      }
      else {
         rc = -EREMOTEIO;
      }

      return rc;
   }

   return 0;
}

/**
 * @brief Translate an HTTP status code into the approprate error code.
 * @return The error or status code
 */
int md_HTTP_status_code_to_error_code( int status_code ) {
   if( status_code == SG_HTTP_TRYAGAIN ) {
      return -EAGAIN;
   }

   if( status_code == 500 ) {
      return -EREMOTEIO;
   }

   if( status_code == 404 ) {
      return -ENOENT;
   }

   return status_code;
}

/**
 * @brief Alloc a download loop
 * @return Pointer to the allocated memory block
 */
struct md_download_loop* md_download_loop_new() {
   return SG_CALLOC( struct md_download_loop, 1 );
}

/**
 * @brief Initialize a download loop
 * @retval 0 Success
 * @retval -ENOMEM Out of Memory
 */
int md_download_loop_init( struct md_download_loop* dlloop, struct md_downloader* dl, int num_downloads ) {

   memset( dlloop, 0, sizeof(struct md_download_loop) );

   dlloop->downloads = SG_CALLOC( struct md_download_context*, num_downloads );
   if( dlloop->downloads == NULL ) {

      return -ENOMEM;
   }

   for( int i = 0; i < num_downloads; i++ ) {
      dlloop->downloads[i] = SG_CALLOC( struct md_download_context, 1 );
      if( dlloop->downloads[i] == NULL ) {

         // clean up
         for( int j = 0; j < i; j++ ) {
            SG_safe_free( dlloop->downloads[i] );
         }

         SG_safe_free( dlloop->downloads );
         return -ENOMEM;
      }
   }

   int rc = md_download_set_init( &dlloop->dlset );
   if( rc != 0 ) {

      // clean up
      for( int j = 0; j < num_downloads; j++ ) {
         SG_safe_free( dlloop->downloads[j] );
      }
      SG_safe_free( dlloop->downloads );
      return rc;
   }

   dlloop->num_downloads = num_downloads;
   dlloop->dl = dl;

   return 0;
}

/**
 * @brief Free a download loop
 * @note Always succeeds
 */
int md_download_loop_free( struct md_download_loop* dlloop ) {

   for( int i = 0; i < dlloop->num_downloads; i++ ) {

      if( !md_download_context_initialized( dlloop->downloads[i] ) ) {
         SG_safe_free( dlloop->downloads[i] );
      }
   }

   SG_safe_free( dlloop->downloads );
   md_download_set_free( &dlloop->dlset );

   memset( dlloop, 0, sizeof(struct md_download_loop) );

   return 0;
}

/**
 * @brief Get the next available download in the download loop.
 * @retval 0 Success, and set *dlctx to point to the available download
 * @retval -EAGAIN No free downloads, and set *dlctx to NULL
 */
int md_download_loop_next( struct md_download_loop* dlloop, struct md_download_context** dlctx ) {

   *dlctx = NULL;
   for( int i = 0; i < dlloop->num_downloads; i++ ) {

      if( !md_download_context_initialized( dlloop->downloads[i] ) ) {

         *dlctx = dlloop->downloads[i];

         return 0;
      }
   }

   return -EAGAIN;
}

/**
 * @brief Have the download loop process the download when it runs
 *
 * md_download_loop_run knows to return if the given download finishes.
 * @return Status of running md_download_set_add
 * @see md_download_set_add
 * @retval 0 Success
 * @retval -ENOMEM Out of Memory
 */
int md_download_loop_watch( struct md_download_loop* dlloop, struct md_download_context* dlctx ) {
   return md_download_set_add( &dlloop->dlset, dlctx );
}

/**
 * @brief Run the download loop, until at least one download completes
 * @retval 0 Success
 * @retval 1 if there are no more downloads
 * @retval -errno on critical failure to wait
 */
int md_download_loop_run( struct md_download_loop* dlloop ) {

   int rc = 0;
   dlloop->started = true;

   while( md_download_set_size( &dlloop->dlset ) > 0 ) {

      // wait for some downloads to finish, but be resillent against deadlock
      rc = md_download_context_wait_any( &dlloop->dlset, 10000 );
      if( rc != 0 && rc != -ETIMEDOUT ) {

         // failed
         SG_error("md_download_context_wait_any(%p) rc = %d\n", &dlloop->dlset, rc );
         return rc;
      }

      else if( rc == -ETIMEDOUT ) {

         SG_debug("still waiting on download set %p\n", &dlloop->dlset );
         continue;
      }

      return rc;
   }

   return 1;
}

/**
 * @brief Find a finished download
 * @note The caller must unref and free when done with it
 * @retval 0 Success, and set *dlctx to point to the finished download
 * @retval -EAGAIN if there are no finished downloads, and set *dlctx to NULL
 */
int md_download_loop_finished( struct md_download_loop* dlloop, struct md_download_context** dlctx ) {

   *dlctx = NULL;
   for( int i = 0; i < dlloop->num_downloads; i++ ) {

      if( md_download_context_initialized( dlloop->downloads[i] ) && md_download_context_finalized( dlloop->downloads[i] ) && md_download_set_contains( &dlloop->dlset, dlloop->downloads[i] ) ) {

         *dlctx = dlloop->downloads[i];
         md_download_set_clear( &dlloop->dlset, *dlctx );

         return 0;
      }
   }

   return -EAGAIN;
}

/**
 * @brief Get the number of running downloads
 * @retval >=0 Number of downloads running
 */
int md_download_loop_num_running( struct md_download_loop* dlloop ) {

   int ret = 0;

   for( int i = 0; i < dlloop->num_downloads; i++ ) {

      if( md_download_context_running( dlloop->downloads[i] ) ) {

         ret++;
      }
   }

   return ret;
}

/**
 * @brief Get the number of initialized downloads
 * @retval >=0 Number of initialized downloads
 */
int md_download_loop_num_initialized( struct md_download_loop* dlloop ) {

   int ret = 0;

   for( int i = 0; i < dlloop->num_downloads; i++ ) {

      if( md_download_context_initialized( dlloop->downloads[i] ) ) {

         ret++;
      }
   }

   return ret;
}

/**
 * @brief Determine if there are downloads outstanding
 * @retval True Outstanding downloads
 * @retval False No outstanding downloads
 */
bool md_download_loop_running( struct md_download_loop* dlloop ) {

   return (dlloop->started && md_download_loop_num_initialized( dlloop ) > 0);
}

/**
 * @brief Cancel all downloads in a download loop, but in a fail-fast manner
 * @retval 0 Success
 * @retval -errno Failure to cancel (see md_download_context_cancel)
 * @see md_download_context_cancel
 */
int md_download_loop_abort( struct md_download_loop* dlloop ) {

   int rc = 0;
   struct md_download_context* dlctx = NULL;

   for( int i = 0; i < dlloop->num_downloads; i++ ) {

      dlctx = dlloop->downloads[i];

      if( !dlctx->initialized ) {
         continue;
      }

      // cancel it
      rc = md_download_context_cancel( dlloop->dl, dlctx );
      if( rc != 0 ) {

         SG_error("md_download_context_cancel( %p ) rc = %d\n", dlctx, rc );
         break;
      }
   }

   return rc;
}

/**
 * @brief Unreference all downloads in a download loop
 *
 * Free them and pass their CURL handles to curl_release if their ref counts reach 0
 * If given, call the curl release function on each CURL handle
 * @note All the download contexts here have to be ref'ed!
 * @note Always succeeds
 */
int md_download_loop_cleanup( struct md_download_loop* dlloop, md_download_curl_release_func curl_release, void* release_cls ) {

   int rc = 0;
   struct md_download_context* dlctx = NULL;
   CURL* curl = NULL;

   SG_debug("Clean up download loop %p (set %p)\n", dlloop, &dlloop->dlset);

   for( int i = 0; i < dlloop->num_downloads; i++ ) {

      dlctx = dlloop->downloads[i];

      if( !dlctx->initialized ) {
         // clear from parent set
         SG_debug("Clear download %p from set %p in loop %p\n", dlctx, &dlloop->dlset, dlloop );
         md_download_set_clear( &dlloop->dlset, dlctx );
         continue;
      }
      else {
         // remove from parent set
         md_download_context_clear_set( dlctx );
      }

      // wait to finish
      SG_debug("Wait for download context %p (set %p, loop %p)\n", dlctx, &dlloop->dlset, dlloop);
      md_download_context_wait( dlctx, -1 );

      // unref
      rc = md_download_context_unref( dlctx );
      if( rc > 0 ) {

         // fully unref'ed
         md_download_context_free( dlctx, &curl );

         if( curl_release != NULL ) {

            (*curl_release)( curl, release_cls );
         }
         else {

            // default behavior
            curl_easy_cleanup( curl );
         }
      }
   }

   return 0;
}

/**
 * @brief Find the next initialized download at the given offset, incrementing *i
 *
 * If i == NULL, just find the first initialized download
 * @return A pointer to the download, Success
 * @retval NULL Attempting to index beyond the array (i >= num_downloads)
 */
struct md_download_context* md_download_loop_next_initialized( struct md_download_loop* dlloop, int* i ) {

   if( i == NULL ) {
      for( int j = 0; j < dlloop->num_downloads; j++ ) {

         if( dlloop->downloads[j]->initialized ) {

            return dlloop->downloads[j];
         }
      }

      return NULL;
   }
   else {

      while( *i < dlloop->num_downloads && !dlloop->downloads[*i]->initialized ) {
         *i = (*i) + 1;
      }

      if( *i >= dlloop->num_downloads ) {
         return NULL;
      }
      else {
         struct md_download_context* ret = dlloop->downloads[*i];
         *i = (*i) + 1;
         return ret;
      }
   }
}

/**
 * @brief Download a single item, synchronously, up to max_size bytes (pass -1 for no maximum size)
 * @retval 0 Success, and populate *buf and *buflen with the downloaded data
 * @retval -ENOMEM Out of Memory
 * @retval -ETIMEDOUT The tranfser could not complete in time
 * @retval -EAGAIN Signaled to retry the request
 * @retval -EREMOTEIO HTTP error is >= 500
 * @retval -499<=rc>=-400 The HTTP error was in the range 400 to 499
 * @retval Other -errno on socket- and recv-related errors
 */
int md_download_run( CURL* curl, off_t max_size, char** buf, off_t* buf_len ) {

   int rc = 0;
   long http_status = 0;
   long os_errno = 0;
   struct md_bound_response_buffer brb;

   // initialize
   rc = md_bound_response_buffer_init( &brb, max_size );
   if( rc != 0 ) {

      return rc;
   }

   // point curl to our brb
   curl_easy_setopt( curl, CURLOPT_WRITEDATA, &brb );
   curl_easy_setopt( curl, CURLOPT_WRITEFUNCTION, md_get_callback_bound_response_buffer );

   rc = curl_easy_perform( curl );

   // get HTTP status and error code
   curl_easy_getinfo( curl, CURLINFO_RESPONSE_CODE, &http_status );
   curl_easy_getinfo( curl, CURLINFO_OS_ERRNO, &os_errno );

   if( rc != 0 || http_status >= 400 ) {

      SG_error("curl_easy_perform rc = %d, HTTP status = %ld, os_errno = %ld\n", rc, http_status, os_errno );

      rc = md_download_interpret_errors( http_status, rc, os_errno );

      // clean up
      md_bound_response_buffer_free( &brb );
      return rc;
   }

   // extract the buffer
   *buf = md_response_buffer_to_string( brb.rb );
   *buf_len = md_response_buffer_size( brb.rb );

   md_bound_response_buffer_free( &brb );

   if( *buf == NULL ) {

      rc = -ENOMEM;
   }

   return rc;
}

/**
 * @brief Initialize a bound response buffer
 * @retval 0 Success
 * @retval -ENOMEM if allocation failed
 */
int md_bound_response_buffer_init( struct md_bound_response_buffer* brb, off_t max_size ) {

   memset( brb, 0, sizeof(struct md_bound_response_buffer) );

   brb->rb = SG_safe_new( md_response_buffer_t() );
   if( brb == NULL ) {
      return -ENOMEM;
   }

   brb->max_size = max_size;
   brb->size = 0;

   return 0;
}

/**
 * @brief Free a bound response buffer
 * @note Always succeeds
 */
int md_bound_response_buffer_free( struct md_bound_response_buffer* brb ) {

   if( brb->rb != NULL ) {
      md_response_buffer_free( brb->rb );
      SG_safe_delete( brb->rb );
   }

   memset( brb, 0, sizeof(struct md_bound_response_buffer) );

   return 0;
}


text:
import mock
from neutron_lib import constants
from neutron_lib import exceptions as n_exc
from oslo_db import exception as db_exc

from neutron.api.rpc.handlers import dhcp_rpc
from neutron.callbacks import resources
from neutron.common import constants as n_const
from neutron.common import exceptions
from neutron.common import utils
from neutron.db import provisioning_blocks
from neutron.extensions import portbindings
from neutron.tests import base

class TestDhcpRpcCallback(base.BaseTestCase):

    def setUp(self):
        super(TestDhcpRpcCallback, self).setUp()
        self.plugin_p = mock.patch('neutron.manager.NeutronManager.get_plugin')
        get_plugin = self.plugin_p.start()
        self.plugin = mock.MagicMock()
        get_plugin.return_value = self.plugin
        self.callbacks = dhcp_rpc.DhcpRpcCallback()
        self.log_p = mock.patch('neutron.api.rpc.handlers.dhcp_rpc.LOG')
        self.log = self.log_p.start()
        set_dirty_p = mock.patch('neutron.quota.resource_registry.'
                                 'set_resources_dirty')
        self.mock_set_dirty = set_dirty_p.start()
        self.utils_p = mock.patch('neutron.plugins.common.utils.create_port')
        self.utils = self.utils_p.start()
        self.segment_p = mock.patch(
            'neutron.manager.NeutronManager.get_service_plugins')
        self.get_service_plugins = self.segment_p.start()
        self.segment_plugin = mock.MagicMock()

    def test_group_by_network_id(self):
        port1 = {'network_id': 'a'}
        port2 = {'network_id': 'b'}
        port3 = {'network_id': 'a'}
        grouped_ports = self.callbacks._group_by_network_id(
                                                        [port1, port2, port3])
        expected = {'a': [port1, port3], 'b': [port2]}
        self.assertEqual(expected, grouped_ports)

    def test_get_active_networks_info(self):
        plugin_retval = [{'id': 'a'}, {'id': 'b'}]
        self.plugin.get_networks.return_value = plugin_retval
        port = {'network_id': 'a'}
        subnet = {'network_id': 'b', 'id': 'c'}
        self.plugin.get_ports.return_value = [port]
        self.plugin.get_subnets.return_value = [subnet]
        networks = self.callbacks.get_active_networks_info(mock.Mock(),
                                                           host='host')
        expected = [{'id': 'a', 'subnets': [], 'ports': [port]},
                    {'id': 'b', 'subnets': [subnet], 'ports': []}]
        self.assertEqual(expected, networks)

    def test_get_active_networks_info_with_routed_networks(self):
        self.get_service_plugins.return_value = {
            'segments': self.segment_plugin
        }
        plugin_retval = [{'id': 'a'}, {'id': 'b'}]
        port = {'network_id': 'a'}
        subnets = [{'network_id': 'b', 'id': 'c', 'segment_id': '1'},
                   {'network_id': 'a', 'id': 'e'},
                   {'network_id': 'b', 'id': 'd', 'segment_id': '3'}]
        self.plugin.get_ports.return_value = [port]
        self.plugin.get_networks.return_value = plugin_retval
        hostseg_retval = ['1', '2']
        self.segment_plugin.get_segments_by_hosts.return_value = hostseg_retval
        self.plugin.get_subnets.return_value = subnets
        networks = self.callbacks.get_active_networks_info(mock.Mock(),
                                                           host='host')
        expected = [{'id': 'a', 'subnets': [subnets[1]], 'ports': [port]},
                    {'id': 'b', 'subnets': [subnets[0]], 'ports': []}]
        self.assertEqual(expected, networks)

    def _test__port_action_with_failures(self, exc=None, action=None):
        port = {
            'network_id': 'foo_network_id',
            'device_owner': constants.DEVICE_OWNER_DHCP,
            'fixed_ips': [{'subnet_id': 'foo_subnet_id'}]
        }
        self.plugin.create_port.side_effect = exc
        self.utils.side_effect = exc
        self.assertIsNone(self.callbacks._port_action(self.plugin,
                                                      mock.Mock(),
                                                      {'port': port},
                                                      action))

    def _test__port_action_good_action(self, action, port, expected_call):
        self.callbacks._port_action(self.plugin, mock.Mock(),
                                    port, action)
        if action == 'create_port':
            self.utils.assert_called_once_with(mock.ANY, mock.ANY, mock.ANY)
        else:
            self.plugin.assert_has_calls([expected_call])

    def test_port_action_create_port(self):
        self._test__port_action_good_action(
            'create_port', mock.Mock(),
            mock.call.create_port(mock.ANY, mock.ANY))

    def test_port_action_update_port(self):
        fake_port = {'id': 'foo_port_id', 'port': mock.Mock()}
        self._test__port_action_good_action(
            'update_port', fake_port,
            mock.call.update_port(mock.ANY, 'foo_port_id', mock.ANY))

    def test__port_action_bad_action(self):
        self.assertRaises(
            n_exc.Invalid,
            self._test__port_action_with_failures,
            exc=None,
            action='foo_action')

    def test_create_port_catch_network_not_found(self):
        self._test__port_action_with_failures(
            exc=n_exc.NetworkNotFound(net_id='foo_network_id'),
            action='create_port')

    def test_create_port_catch_subnet_not_found(self):
        self._test__port_action_with_failures(
            exc=n_exc.SubnetNotFound(subnet_id='foo_subnet_id'),
            action='create_port')

    def test_create_port_catch_db_reference_error(self):
        self._test__port_action_with_failures(
            exc=db_exc.DBReferenceError('a', 'b', 'c', 'd'),
            action='create_port')

    def test_create_port_catch_ip_generation_failure_reraise(self):
        self.assertRaises(
            n_exc.IpAddressGenerationFailure,
            self._test__port_action_with_failures,
            exc=n_exc.IpAddressGenerationFailure(net_id='foo_network_id'),
            action='create_port')

    def test_create_port_catch_and_handle_ip_generation_failure(self):
        self.plugin.get_subnet.side_effect = (
            n_exc.SubnetNotFound(subnet_id='foo_subnet_id'))
        self._test__port_action_with_failures(
            exc=n_exc.IpAddressGenerationFailure(net_id='foo_network_id'),
            action='create_port')
        self._test__port_action_with_failures(
            exc=n_exc.InvalidInput(error_message='sorry'),
            action='create_port')

    def test_update_port_missing_port_on_get(self):
        self.plugin.get_port.side_effect = n_exc.PortNotFound(port_id='66')
        self.assertIsNone(self.callbacks.update_dhcp_port(
            context='ctx', host='host', port_id='66',
            port={'port': {'network_id': 'a'}}))

    def test_update_port_missing_port_on_update(self):
        self.plugin.get_port.return_value = {
            'device_id': n_const.DEVICE_ID_RESERVED_DHCP_PORT}
        self.plugin.update_port.side_effect = n_exc.PortNotFound(port_id='66')
        self.assertIsNone(self.callbacks.update_dhcp_port(
            context='ctx', host='host', port_id='66',
            port={'port': {'network_id': 'a'}}))

    def test_get_network_info_return_none_on_not_found(self):
        self.plugin.get_network.side_effect = n_exc.NetworkNotFound(net_id='a')
        retval = self.callbacks.get_network_info(mock.Mock(), network_id='a')
        self.assertIsNone(retval)

    def _test_get_network_info(self, segmented_network=False,
                               routed_network=False):
        network_retval = dict(id='a')
        if not routed_network:
            subnet_retval = [dict(id='a'), dict(id='c'), dict(id='b')]
        else:
            subnet_retval = [dict(id='c', segment_id='1'),
                             dict(id='a', segment_id='1')]
        port_retval = mock.Mock()

        self.plugin.get_network.return_value = network_retval
        self.plugin.get_subnets.return_value = subnet_retval
        self.plugin.get_ports.return_value = port_retval
        if segmented_network:
            self.segment_plugin.get_segments.return_value = [dict(id='1'),
                                                             dict(id='2')]
            self.segment_plugin.get_segments_by_hosts.return_value = ['1']

        retval = self.callbacks.get_network_info(mock.Mock(), network_id='a')
        self.assertEqual(retval, network_retval)
        if not routed_network:
            sorted_subnet_retval = [dict(id='a'), dict(id='b'), dict(id='c')]
        else:
            sorted_subnet_retval = [dict(id='a', segment_id='1'),
                                    dict(id='c', segment_id='1')]
        self.assertEqual(retval['subnets'], sorted_subnet_retval)
        self.assertEqual(retval['ports'], port_retval)

    def test_get_network_info(self):
        self._test_get_network_info()

    def test_get_network_info_with_routed_network(self):
        self.get_service_plugins.return_value = {
            'segments': self.segment_plugin
        }
        self._test_get_network_info(segmented_network=True,
                                    routed_network=True)

    def test_get_network_info_with_segmented_network_but_not_routed(self):
        self.get_service_plugins.return_value = {
            'segments': self.segment_plugin
        }
        self._test_get_network_info(segmented_network=True)

    def test_get_network_info_with_non_segmented_network(self):
        self.get_service_plugins.return_value = {
            'segments': self.segment_plugin
        }
        self._test_get_network_info()

    def test_update_dhcp_port_verify_port_action_port_dict(self):
        port = {'port': {'network_id': 'foo_network_id',
                         'device_owner': constants.DEVICE_OWNER_DHCP,
                         'fixed_ips': [{'subnet_id': 'foo_subnet_id'}]}
                }
        expected_port = {'port': {'network_id': 'foo_network_id',
                                  'device_owner': constants.DEVICE_OWNER_DHCP,
                                  portbindings.HOST_ID: 'foo_host',
                                  'fixed_ips': [{'subnet_id': 'foo_subnet_id'}]
                                  },
                         'id': 'foo_port_id'
                         }

        def _fake_port_action(plugin, context, port, action):
            self.assertEqual(expected_port, port)

        self.plugin.get_port.return_value = {
            'device_id': n_const.DEVICE_ID_RESERVED_DHCP_PORT}
        self.callbacks._port_action = _fake_port_action
        self.callbacks.update_dhcp_port(mock.Mock(),
                                        host='foo_host',
                                        port_id='foo_port_id',
                                        port=port)

    def test_update_reserved_dhcp_port(self):
        port = {'port': {'network_id': 'foo_network_id',
                         'device_owner': constants.DEVICE_OWNER_DHCP,
                         'fixed_ips': [{'subnet_id': 'foo_subnet_id'}]}
                }
        expected_port = {'port': {'network_id': 'foo_network_id',
                                  'device_owner': constants.DEVICE_OWNER_DHCP,
                                  portbindings.HOST_ID: 'foo_host',
                                  'fixed_ips': [{'subnet_id': 'foo_subnet_id'}]
                                  },
                         'id': 'foo_port_id'
                         }

        def _fake_port_action(plugin, context, port, action):
            self.assertEqual(expected_port, port)

        self.plugin.get_port.return_value = {
            'device_id': utils.get_dhcp_agent_device_id('foo_network_id',
                                                        'foo_host')}
        self.callbacks._port_action = _fake_port_action
        self.callbacks.update_dhcp_port(
            mock.Mock(), host='foo_host', port_id='foo_port_id', port=port)

        self.plugin.get_port.return_value = {
            'device_id': 'other_id'}
        self.assertRaises(exceptions.DhcpPortInUse,
                          self.callbacks.update_dhcp_port,
                          mock.Mock(),
                          host='foo_host',
                          port_id='foo_port_id',
                          port=port)

    def test_update_dhcp_port(self):
        port = {'port': {'network_id': 'foo_network_id',
                         'device_owner': constants.DEVICE_OWNER_DHCP,
                         'fixed_ips': [{'subnet_id': 'foo_subnet_id'}]}
                }
        expected_port = {'port': {'network_id': 'foo_network_id',
                                  'device_owner': constants.DEVICE_OWNER_DHCP,
                                  portbindings.HOST_ID: 'foo_host',
                                  'fixed_ips': [{'subnet_id': 'foo_subnet_id'}]
                                  },
                         'id': 'foo_port_id'
                         }
        self.plugin.get_port.return_value = {
            'device_id': n_const.DEVICE_ID_RESERVED_DHCP_PORT}
        self.callbacks.update_dhcp_port(mock.Mock(),
                                        host='foo_host',
                                        port_id='foo_port_id',
                                        port=port)
        self.plugin.assert_has_calls([
            mock.call.update_port(mock.ANY, 'foo_port_id', expected_port)])

    def test_release_dhcp_port(self):
        port_retval = dict(id='port_id', fixed_ips=[dict(subnet_id='a')])
        self.plugin.get_ports.return_value = [port_retval]

        self.callbacks.release_dhcp_port(mock.ANY, network_id='netid',
                                         device_id='devid')

        self.plugin.assert_has_calls([
            mock.call.delete_ports_by_device_id(mock.ANY, 'devid', 'netid')])

    def test_dhcp_ready_on_ports(self):
        context = mock.Mock()
        port_ids = range(10)
        with mock.patch.object(provisioning_blocks,
                               'provisioning_complete') as pc:
            self.callbacks.dhcp_ready_on_ports(context, port_ids)
        calls = [mock.call(context, port_id, resources.PORT,
                           provisioning_blocks.DHCP_ENTITY)
                 for port_id in port_ids]
        pc.assert_has_calls(calls)

redpajama_set_name: RedPajamaGithub


text:
package uk.co.turingatemyhamster
package owl2

/**
 * An abstraction of: http://www.w3.org/TR/2012/REC-owl2-syntax-20121211/#Annotations
 *
 * @author Matthew Pocock
 */
trait AnnotationsModule {

  importedModules : owl2.IriModule with owl2.EntitiesLiteralsAnonymousIndividualsModule =>

  type AnnotationValue <: AnyRef

  type AnonymousIndividual <: AnnotationValue

  type IRI <: AnnotationValue

  type Literal <: AnnotationValue

}


text:
TOKEN=$(keystone token-get | grep -w id  | awk -F"|" '{print $3}')
echo $TOKEN
#TOKEN="3ac866661cac476a903af4b6f90c9773"
curl -i 'http://127.0.0.1:9696/v2.0/vm/devices.json' -X POST -H "X-Auth-Token:${TOKEN}" -H "Content-Type: application/json" -H "Accept: application/json" -H "User-Agent: python-neutronclient" -d '{
    "device": {
        "template_id": "855d2fc6-fed8-44fc-ac74-a90c90fd9329",
        "auth":[{"username":"test1111", "password":"123456"}],
        "attributes": {
            "fixed_ips": [
                {
                    "subnet_id": "b82b6675-4040-4ebd-bc06-f762753b2811",
                    "gateway_ip":"True"
                }
            ]
        }
    }
}'

        #"template_id": "8316d5e9-8de3-4d09-b6a9-29b299cf7c07",
                    #"subnet_id": "77fbe5b9-13cc-44f5-b37c-998683767319"
#name='servicevm_mgmt_security_group'
#service_tenant=$(keystone tenant-list | grep -w services  | cut -d'|' -f2)
#echo $service_tenant
#security_group=$(neutron security-group-create --tenant-id $service_tenant  $name |grep ' id ' | awk '{print $2}'
#echo $security_group
#neutron security-group-rule-create $security_group --tenant-id $service_tenant --direction egress --ethertype IPv4 --protocol tcp --port-range-min 80 --port-range-max 80

#name='servicevm_internal_port_security_group'
#service_tenant=$(keystone tenant-list | grep -w services  | cut -d'|' -f2)
#echo $service_tenant
#security_group=$(neutron security-group-create --tenant-id $service_tenant  $name |grep ' id ' | awk '{print $2}')
#echo $security_group
#neutron security-group-rule-create $security_group --tenant-id $service_tenant --direction egress --ethertype IPv4


text:
package org.zaproxy.zap.extension.help;

import java.util.Locale;

import javax.help.HelpSet;
import javax.help.Map.ID;
import javax.help.TOCItem;

/**
 * A {@code TOCItem} that has the value of "tocid" attribute of a "tocitem" element.
 *
 * @see TOCItem
 * @see ZapTocMerger
 * @see ZapTocView
 */
public class ZapTocItem extends TOCItem {

    private static final long serialVersionUID = -1267310421450082382L;

    private final String tocId;

    public ZapTocItem() {
        super(null, null, null);

        this.tocId = null;
    }

    public ZapTocItem(ID id, ID imageID, Locale locale, String tocId) {
        super(id, imageID, locale);

        this.tocId = tocId;
    }

    public ZapTocItem(ID id, ID imageID, HelpSet hs, Locale locale, String tocId) {
        super(id, imageID, hs, locale);

        this.tocId = tocId;
    }

    public String getTocId() {
        return tocId;
    }
}


text:
 #  Copyright 2008-2010 Amazon Technologies, Inc
 #  Licensed under the Apache License, Version 2.0 (the "License");
#
#  You may not use this file except in compliance with the License.
#  You may obtain a copy of the License at: http://aws.amazon.com/apache2.0
#  This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
#  CONDITIONS OF ANY KIND, either express or implied. See the License for the
#  specific language governing permissions and limitations under the License.
##############################################################################
require 'base64'
require 'cgi'
require 'openssl'

module Amazon
  module FPS

    #
    # Copyright:: Copyright (c) 2009 Amazon.com, Inc. or its affiliates.  All Rights Reserved.
    #
    # RFC 2104-compliant HMAC signature for request parameters
    #  Implements AWS Signature, as per following spec:
    #
    # If Signature Version is 1, it performs the following:
    #
    # Sorts all  parameters (including SignatureVersion and excluding Signature,
    # the value of which is being created), ignoring case.
    #
    # Iterate over the sorted list and append the parameter name (in original case)
    # and then its value. It will not URL-encode the parameter values before
    # constructing this string. There are no separators.
    #
    # If Signature Version is 2, string to sign is based on following:
    #
    #    1. The HTTP Request Method followed by an ASCII newline (%0A)
    #    2. The HTTP Host header in the form of lowercase host, followed by an ASCII newline.
    #    3. The URL encoded HTTP absolute path component of the URI
    #       (up to but not including the query string parameters);
    #       if this is empty use a forward '/'. This parameter is followed by an ASCII newline.
    #    4. The concatenation of all query string components (names and values)
    #       as UTF-8 characters which are URL encoded as per RFC 3986
    #       (hex characters MUST be uppercase), sorted using lexicographic byte ordering.
    #       Parameter names are separated from their values by the '=' character
    #       (ASCII character 61), even if the value is empty.
    #       Pairs of parameter and values are separated by the '&' character (ASCII code 38).
    #
    class SignatureUtils

      SIGNATURE_KEYNAME = "Signature"
      SIGNATURE_METHOD_KEYNAME = "SignatureMethod"
      SIGNATURE_VERSION_KEYNAME = "SignatureVersion"

      HMAC_SHA256_ALGORITHM = "HmacSHA256"
      HMAC_SHA1_ALGORITHM = "HmacSHA1"

      def self.sign_parameters(args)
        signature_version = args[:parameters][SIGNATURE_VERSION_KEYNAME]
        string_to_sign = "";
        algorithm = 'sha1';
        if (signature_version == '1') then
          string_to_sign = calculate_string_to_sign_v1(args)
        elsif (signature_version == '2') then
          algorithm = get_algorithm(args[:parameters][SIGNATURE_METHOD_KEYNAME])
          string_to_sign = calculate_string_to_sign_v2(args)
        else
          raise "Invalid Signature Version specified"
        end
        return compute_signature(string_to_sign, args[:aws_secret_key], algorithm)
      end

      # Convert a string into URL encoded form.
      def self.urlencode(plaintext)
        CGI.escape(plaintext.to_s).gsub("+", "%20").gsub("%7E", "~")
      end

      private # All the methods below are private

      def self.calculate_string_to_sign_v1(args)
        parameters = args[:parameters]

        # exclude any existing Signature parameter from the canonical string
        sorted = (parameters.reject { |k, v| k == SIGNATURE_KEYNAME }).sort { |a,b| a[0].downcase <=> b[0].downcase }

        canonical = ''
        sorted.each do |v|
          canonical << v[0]
          canonical << v[1] unless(v[1].nil?)
        end

        return canonical
      end

      def self.calculate_string_to_sign_v2(args)
        parameters = args[:parameters]

        uri = args[:uri]
        uri = "/" if uri.nil? or uri.empty?
        uri = urlencode(uri).gsub("%2F", "/")

        verb = args[:verb]
        host = args[:host].downcase

        # exclude any existing Signature parameter from the canonical string
        sorted = (parameters.reject { |k, v| k == SIGNATURE_KEYNAME }).sort

        canonical = "#{verb}\n#{host}\n#{uri}\n"
        isFirst = true

        sorted.each { |v|
          if(isFirst) then
            isFirst = false
          else
            canonical << '&'
          end

          canonical << urlencode(v[0])
          unless(v[1].nil?) then
            canonical << '='
            canonical << urlencode(v[1])
          end
        }

        return canonical
      end

      def self.get_algorithm(signature_method)
        return 'sha256' if (signature_method == HMAC_SHA256_ALGORITHM);
        return 'sha1'
      end

      def self.compute_signature(canonical, aws_secret_key, algorithm = 'sha1')
        digest = OpenSSL::Digest::Digest.new(algorithm)
        return Base64.encode64(OpenSSL::HMAC.digest(digest, aws_secret_key, canonical)).chomp
      end

    end

  end
end


text:
<?xml version="1.0" encoding="UTF-8" ?>
<class name="Resource" inherits="Reference" category="Core" version="3.2">
	<brief_description>
		Base class for all resources.
	</brief_description>
	<description>
		Resource is the base class for all Godot-specific resource types, serving primarily as data containers. They are reference counted and freed when no longer in use. They are also cached once loaded from disk, so that any further attempts to load a resource from a given path will return the same reference (all this in contrast to a [Node], which is not reference counted and can be instanced from disk as many times as desired). Resources can be saved externally on disk or bundled into another object, such as a [Node] or another resource.
	</description>
	<tutorials>
		<link>https://docs.godotengine.org/en/latest/getting_started/step_by_step/resources.html</link>
	</tutorials>
	<methods>
		<method name="_setup_local_to_scene" qualifiers="virtual">
			<return type="void">
			</return>
			<description>
				Virtual function which can be overridden to customize the behavior value of [method setup_local_to_scene].
			</description>
		</method>
		<method name="duplicate" qualifiers="const">
			<return type="Resource">
			</return>
			<argument index="0" name="subresources" type="bool" default="false">
			</argument>
			<description>
				Duplicates the resource, returning a new resource. By default, sub-resources are shared between resource copies for efficiency, this can be changed by passing [code]true[/code] to the [code]subresources[/code] argument.
			</description>
		</method>
		<method name="get_local_scene" qualifiers="const">
			<return type="Node">
			</return>
			<description>
				If [member resource_local_to_scene] is enabled and the resource was loaded from a [PackedScene] instantiation, returns the local scene where this resource's unique copy is in use. Otherwise, returns [code]null[/code].
			</description>
		</method>
		<method name="get_rid" qualifiers="const">
			<return type="RID">
			</return>
			<description>
				Returns the RID of the resource (or an empty RID). Many resources (such as [Texture], [Mesh], etc) are high-level abstractions of resources stored in a server, so this function will return the original RID.
			</description>
		</method>
		<method name="setup_local_to_scene">
			<return type="void">
			</return>
			<description>
				This method is called when a resource with [member resource_local_to_scene] enabled is loaded from a [PackedScene] instantiation. Its behavior can be customized by overriding [method _setup_local_to_scene] from script.
				For most resources, this method performs no base logic. [ViewportTexture] performs custom logic to properly set the proxy texture and flags in the local viewport.
			</description>
		</method>
		<method name="take_over_path">
			<return type="void">
			</return>
			<argument index="0" name="path" type="String">
			</argument>
			<description>
				Sets the path of the resource, potentially overriding an existing cache entry for this path. This differs from setting [member resource_path], as the latter would error out if another resource was already cached for the given path.
			</description>
		</method>
	</methods>
	<members>
		<member name="resource_local_to_scene" type="bool" setter="set_local_to_scene" getter="is_local_to_scene" default="false">
			If [code]true[/code], the resource will be made unique in each instance of its local scene. It can thus be modified in a scene instance without impacting other instances of that same scene.
		</member>
		<member name="resource_name" type="String" setter="set_name" getter="get_name" default="""">
			The name of the resource. This is an optional identifier.
		</member>
		<member name="resource_path" type="String" setter="set_path" getter="get_path" default="""">
			The path to the resource. In case it has its own file, it will return its filepath. If it's tied to the scene, it will return the scene's path, followed by the resource's index.
		</member>
	</members>
	<signals>
		<signal name="changed">
			<description>
				Emitted whenever the resource changes.
			</description>
		</signal>
	</signals>
	<constants>
	</constants>
</class>


text:
ACCEPTED

#### According to
International Plant Names Index

#### Published in
null

#### Original name
Candollea pilosa Labill.

### Remarks
null


text:
package com.netflix.spinnaker.orca.clouddriver.pipeline.manifest;

import com.netflix.spinnaker.orca.clouddriver.tasks.MonitorKatoTask;
import com.netflix.spinnaker.orca.clouddriver.tasks.manifest.DeleteManifestTask;
import com.netflix.spinnaker.orca.clouddriver.tasks.manifest.ManifestForceCacheRefreshTask;
import com.netflix.spinnaker.orca.clouddriver.tasks.manifest.PromoteManifestKatoOutputsTask;
import com.netflix.spinnaker.orca.clouddriver.tasks.manifest.ResolveTargetManifestTask;
import com.netflix.spinnaker.orca.pipeline.StageDefinitionBuilder;
import com.netflix.spinnaker.orca.pipeline.TaskNode;
import com.netflix.spinnaker.orca.pipeline.model.Stage;
import org.springframework.stereotype.Component;

@Component
public class DeleteManifestStage implements StageDefinitionBuilder {
  public static final String PIPELINE_CONFIG_TYPE = "deleteManifest";

  @Override
  public void taskGraph(Stage stage, TaskNode.Builder builder) {
    builder
        .withTask(ResolveTargetManifestTask.TASK_NAME, ResolveTargetManifestTask.class)
        .withTask(DeleteManifestTask.TASK_NAME, DeleteManifestTask.class)
        .withTask("monitorDelete", MonitorKatoTask.class)
        .withTask(PromoteManifestKatoOutputsTask.TASK_NAME, PromoteManifestKatoOutputsTask.class)
        .withTask(ManifestForceCacheRefreshTask.TASK_NAME, ManifestForceCacheRefreshTask.class);
  }
}


text:
const router = require('express').Router(),
    auth = require('../authorization/passport-config'),
    Album = require('../album/model');

router.get('/', auth.isLogged, (req, res) => {
    Album.find({}, (err, _albums) => {
        err ? res.render('ui/error.ejs', {error: err}) : res.render('ui/index.ejs', {albums: _albums});
    });
});

module.exports = router;


text:
package br.edu.ifc.concordia.in.calculadora;

import java.util.Scanner;

public class Main {
	public static void main (String[] args){
		Scanner scan = new Scanner(System.in);

		System.out.println("Digite a operação desejada: seja soma, subtração, multiplicação ou divisão");
		int operacao = scan.nextInt();
		System.out.println ("digite o número A:");
		double a = scan.nextDouble();
		System.out.println ("DIgite o numero B:");
		double b = scan.nextDouble();

		Calculadora calc = new Calculadora ();
		double resultado = 0;
		if (operacao ==1){
			resultado = calc.soma(a, b);
		} else if (operacao ==2){
			resultado = calc.multiplicacao(a, b);
		}else if (operacao ==3){
			resultado = calc.subtracao(a, b);
		}else if (operacao ==4){
		resultado = calc.divisao(a, b);
		}

		System.out.printf("Resultado = %f\n", String.valueOf(resultado));
		scan.close();
	}
}


text:
layout: post
title:  "Variance"
date:   2016-03-29 22:44:01
categories: statistics
---

Measures the variance of a series of values around their mean value.

$$ Var(X)=\frac{1}{N}\sum_i{(X_i-\bar{X})^2} $$

where $$ \bar{X}=\frac{1}{N}*\sum_i{X_i} $$

The variance can be computed incrementally:

$$ Var(X)=\bar{X^2}-(\bar{X}*\bar{X}) $$

Thus, we sum all $$X_i$$ and all $$X_i^2$$, then build the difference of their
mean values.


text:
require "pa_ruby_tools/version"
require "pa_ruby_tools/do_gem_release"
require "pa_ruby_tools/config_file"

module PaRubyTools
  # Your code goes here...
end


text:
from pprint import pprint

from pprint_data import data

print('PRINT:')
print(data)
print()
print('PPRINT:')
pprint(data)


text:
<?php

namespace Maildocker;

class Mail
{
    public function __construct()
    {
        $this->to = array();
        $this->cc = array();
        $this->bcc = array();
        $this->images = array();
        $this->attachments = array();
        $this->merge_vars = array();
    }

    public function set_from($from_email, $from_name = null)
    {
        $this->from = array('email' => $from_email);
        if($from_name) $this->from['name'] = $from_name;
        return $this;
    }

    protected function add_mail($field, $mail, $name = null, $merge_vars)
    {
        if(is_string($mail))
        {
            $mail = array('email' => $mail);
            if($name) $mail['name'] = $name;
            $this->{$field}[] = $mail;
        }
        elseif(is_array($mail))
        {
            foreach($mail as $email) $this->add_mail($field, $email);
        }
    }

    public function add_to($to, $name = null, $merge_vars = null)
    {
        $this->add_mail('to', $to, $name, $merge_vars);
        return $this;
    }

    public function add_cc($cc, $name = null, $merge_vars = null)
    {
        $this->add_mail('cc', $cc, $name, $merge_vars);
        return $this;
    }

    public function add_bcc($bcc, $name = null, $merge_vars = null)
    {
        $this->add_mail('bcc', $bcc, $name, $merge_vars);
        return $this;
    }

    public function set_subject($subject)
    {
        $this->subject = $subject;
        return $this;
    }

    public function set_template($template)
    {
        $this->template = $template;
        return $this;
    }

    public function set_text($text)
    {
        $this->text = $text;
        return $this;
    }

    public function set_html($html)
    {
        $this->html = $html;
        return $this;
    }

    public function set_replyto($replyto)
    {
        $this->reply_to = $replyto;
        return $this;
    }

    public function set_date($date)
    {
        $this->date = $date;
        return $this;
    }

    public function add_vars($merge_vars)
    {
        $this->merge_vars = array_merge($this->merge_vars, $merge_vars);
        return $this;
    }

    public function set_headers($headers)
    {
        $this->headers = is_string($headers) ? $headers : json_encode($headers);
        return $this;
    }

    protected function add_file($field, $file)
    {
        if(is_array($file)) $this->{$field}[] = $file;
        else
        {
            $handle = fopen($file, 'rb');
            $this->{$field}[] = array(
                'name' => basename($file),
                'type' => mime_content_type($file),
                'content' => base64_encode(fread($handle, filesize($file)))
            );
        }
    }

    public function add_attachment($attachment)
    {
        if(!is_array($attachment)) $attachment = array($attachment);
        foreach($attachment as $file) $this->add_file('attachments', $file);
        return $this;
    }

    public function add_image($image)
    {
        if(!is_array($image)) $image = array($image);
        foreach($image as $file) $this->add_file('images', $file);
        return $this;
    }
}