Subversion Repositories SmartDukaan

//  Copyright (c) 2007-2008 Facebook

//

//  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://www.apache.org/licenses/LICENSE-2.0

//

//  Unless required by applicable law or agreed to in writing, software

//  distributed under the License 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.

//

// See accompanying file LICENSE or visit the Scribe site at:

// http://developers.facebook.com/scribe/

//

// @author Bobby Johnson

// @author James Wang

// @author Jason Sobel

// @author Avinash Lakshman

// @author Anthony Giardullo

#include "common.h"

#include "scribe_server.h"

using namespace apache::thrift;

using namespace apache::thrift::protocol;

using namespace apache::thrift::transport;

using namespace apache::thrift::server;

using namespace apache::thrift::concurrency;

using namespace facebook::fb303;

using namespace scribe::thrift;

using namespace std;

using boost::shared_ptr;

shared_ptr<scribeHandler> g_Handler;

#define DEFAULT_CHECK_PERIOD       5

#define DEFAULT_MAX_MSG_PER_SECOND 100000

#define DEFAULT_MAX_QUEUE_SIZE     5000000

#define DEFAULT_SERVER_THREADS     3

void print_usage(const char* program_name) {

  cout << "Usage: " << program_name << " [-p port] [-c config_file]" << endl;

}

int main(int argc, char **argv) {

  try {

    /* Increase number of fds */

    struct rlimit r_fd = {65535,65535};

    if (-1 == setrlimit(RLIMIT_NOFILE, &r_fd)) {

      LOG_OPER("setrlimit error (setting max fd size)");

    }

    int next_option;

    const char* const short_options = "hp:c:";

    const struct option long_options[] = {

      { "help",   0, NULL, 'h' },

      { "port",   0, NULL, 'p' },

      { "config", 0, NULL, 'c' },

      { NULL,     0, NULL, 'o' },

    };

    unsigned long int port = 0;  // this can also be specified in the conf file, which overrides the command line

    std::string config_file;

    while (0 < (next_option = getopt_long(argc, argv, short_options, long_options, NULL))) {

      switch (next_option) {

      default:

      case 'h':

        print_usage(argv[0]);

        exit(0);

      case 'c':

        config_file = optarg;

        break;

      case 'p':

        port = strtoul(optarg, NULL, 0);

        break;

      }

    }

    // assume a non-option arg is a config file name

    if (optind < argc && config_file.empty()) {

      config_file = argv[optind];

    }

    // seed random number generation with something reasonably unique

    srand(time(NULL) ^ getpid());

    g_Handler = shared_ptr<scribeHandler>(new scribeHandler(port, config_file));

    g_Handler->initialize();

    shared_ptr<TProcessor> processor(new scribeProcessor(g_Handler));

    /* This factory is for binary compatibility. */

    shared_ptr<TProtocolFactory>

      binaryProtocolFactory(new TBinaryProtocolFactory(0, 0, false, false));

    shared_ptr<ThreadManager> thread_manager;

    if (g_Handler->numThriftServerThreads > 1) {

      // create a ThreadManager to process incoming calls

      thread_manager = ThreadManager::

        newSimpleThreadManager(g_Handler->numThriftServerThreads);

      shared_ptr<PosixThreadFactory> thread_factory(new PosixThreadFactory());

      thread_manager->threadFactory(thread_factory);

      thread_manager->start();

    }

    TNonblockingServer server(processor, binaryProtocolFactory,

                              g_Handler->port, thread_manager);

    LOG_OPER("Starting scribe server on port %lu", g_Handler->port);

    fflush(stderr);

    server.serve();

  } catch(std::exception const& e) {

    LOG_OPER("Exception in main: %s", e.what());

  }

  LOG_OPER("scribe server exiting");

  return 0;

}

scribeHandler::scribeHandler(unsigned long int server_port, const std::string& config_file)

  : FacebookBase("Scribe"),

    port(server_port),

    numThriftServerThreads(DEFAULT_SERVER_THREADS),

    checkPeriod(DEFAULT_CHECK_PERIOD),

    pcategories(NULL),

    pcategory_prefixes(NULL),

    configFilename(config_file),

    status(STARTING),

    statusDetails("initial state"),

    numMsgLastSecond(0),

    maxMsgPerSecond(DEFAULT_MAX_MSG_PER_SECOND),

    maxQueueSize(DEFAULT_MAX_QUEUE_SIZE),

    newThreadPerCategory(true) {

  time(&lastMsgTime);

}

scribeHandler::~scribeHandler() {

  deleteCategoryMap(pcategories);

  if (pcategory_prefixes) {

    delete pcategory_prefixes;

    pcategory_prefixes = NULL;

  }

}

// Returns the handler status, but overwrites it with WARNING if it's

// ALIVE and at least one store has a nonempty status.

fb_status scribeHandler::getStatus() {

  RWGuard monitor(scribeHandlerLock);

  Guard status_monitor(statusLock);

  fb_status return_status(status);

  if (status == ALIVE) {

    for (category_map_t::iterator cat_iter = pcategories->begin();

        cat_iter != pcategories->end();

        ++cat_iter) {

      for (store_list_t::iterator store_iter = cat_iter->second->begin();

           store_iter != cat_iter->second->end();

           ++store_iter)

      {

        if (!(*store_iter)->getStatus().empty())

        {

          return_status = WARNING;

          return return_status;

        }

      } // for each store

    } // for each category

  } // if we don't have an interesting top level status

  return return_status;

}

void scribeHandler::setStatus(fb_status new_status) {

  LOG_OPER("STATUS: %s", statusAsString(new_status));

  Guard status_monitor(statusLock);

  status = new_status;

}

// Returns the handler status details if non-empty,

// otherwise the first non-empty store status found

void scribeHandler::getStatusDetails(std::string& _return) {

  RWGuard monitor(scribeHandlerLock);

  Guard status_monitor(statusLock);

  _return = statusDetails;

  if (_return.empty()) {

    if (pcategories) {

      for (category_map_t::iterator cat_iter = pcategories->begin();

          cat_iter != pcategories->end();

          ++cat_iter) {

        for (store_list_t::iterator store_iter = cat_iter->second->begin();

            store_iter != cat_iter->second->end();

            ++store_iter) {

          if (!(_return = (*store_iter)->getStatus()).empty()) {

            return;

          }

        } // for each store

      } // for each category

    }

  } // if we don't have an interesting top level status

  return;

}

void scribeHandler::setStatusDetails(const string& new_status_details) {

  LOG_OPER("STATUS: %s", new_status_details.c_str());

  Guard status_monitor(statusLock);

  statusDetails = new_status_details;

}

const char* scribeHandler::statusAsString(fb_status status) {

  switch (status) {

  case DEAD:

    return "DEAD";

  case STARTING:

    return "STARTING";

  case ALIVE:

    return "ALIVE";

  case STOPPING:

    return "STOPPING";

  case STOPPED:

    return "STOPPED";

  case WARNING:

    return "WARNING";

  default:

    return "unknown status code";

  }

}

// Should be called while holding a writeLock on scribeHandlerLock

bool scribeHandler::createCategoryFromModel(

  const string &category, const boost::shared_ptr<StoreQueue> &model) {

  if ((pcategories == NULL) ||

      (pcategories->find(category) != pcategories->end())) {

    return false;

  }

  LOG_OPER("[%s] Creating new category from model %s", category.c_str(),

           model->getCategoryHandled().c_str());

  // Make sure the category name is sane.

  try {

    string clean_path = boost::filesystem::path(category).string();

    if (clean_path.compare(category) != 0) {

      LOG_OPER("Category not a valid boost filename");

      return false;

    }

  } catch(std::exception const& e) {

    LOG_OPER("Category not a valid boost filename.  Boost exception:%s", e.what());

    return false;

  }

  shared_ptr<StoreQueue> pstore;

  if (newThreadPerCategory) {

    // Create a new thread/StoreQueue for this category

    pstore = shared_ptr<StoreQueue>(new StoreQueue(model, category));

    // queue a command to the store to open it

    pstore->open();

  } else {

    // Use existing StoreQueue

    pstore = model;

  }

  shared_ptr<store_list_t> pstores =

    shared_ptr<store_list_t>(new store_list_t);

  (*pcategories)[category] = pstores;

  pstores->push_back(pstore);

  return true;

}

// Check if we need to deny this request due to throttling

bool scribeHandler::throttleRequest(const vector<LogEntry>&  messages) {

  // Check if we need to rate limit

  if (throttleDeny(messages.size())) {

    incrementCounter("denied for rate");

    return true;

  }

  if (!pcategories || !pcategory_prefixes) {

    // don't bother to spam anything for this, our status should already

    // be showing up as WARNING in the monitoring tools.

    incrementCounter("invalid requests");

    return true;

  }

  // Throttle based on store queues getting too long.

  // Note that there's one decision for all categories, because the whole array passed to us

  // must either succeed or fail together. Checking before we've queued anything also has

  // the nice property that any size array will succeed if we're unloaded before attempting

  // it, so we won't hit a case where there's a client request that will never succeed.

  // Also note that we always check all categories, not just the ones in this request.

  // This is a simplification based on the assumption that most Log() calls contain most

  // categories.

  unsigned long max_count = 0;

  for (category_map_t::iterator cat_iter = pcategories->begin();

       cat_iter != pcategories->end();

       ++cat_iter) {

    shared_ptr<store_list_t> pstores = cat_iter->second;

    if (!pstores) {

      throw std::logic_error("throttle check: iterator in category map holds null pointer");

    }

    for (store_list_t::iterator store_iter = pstores->begin();

         store_iter != pstores->end();

         ++store_iter) {

      if (*store_iter == NULL) {

        throw std::logic_error("throttle check: iterator in store map holds null pointer");

      } else {

        unsigned long size = (*store_iter)->getSize();

        if (size > max_count) {

          max_count = size;

        }

      }

    }

  }

  if (max_count > maxQueueSize) {

    incrementCounter("denied for queue size");

    return true;

  }

  return false;

}

// Should be called while holding a writeLock on scribeHandlerLock

shared_ptr<store_list_t> scribeHandler::createNewCategory(

  const string& category) {

  shared_ptr<store_list_t> store_list;

  // First, check the list of category prefixes for a model

  category_prefix_map_t::iterator cat_prefix_iter = pcategory_prefixes->begin();

  while (cat_prefix_iter != pcategory_prefixes->end()) {

    string::size_type len = cat_prefix_iter->first.size();

    if (cat_prefix_iter->first.compare(0, len-1, category, 0, len-1) == 0) {

      // Found a matching prefix model

      createCategoryFromModel(category, cat_prefix_iter->second);

      category_map_t::iterator cat_iter = pcategories->find(category);

      if (cat_iter != pcategories->end()) {

        store_list = cat_iter->second;

      } else {

        LOG_OPER("failed to create new prefix store for category <%s>",

                 category.c_str());

      }

      break;

    }

    cat_prefix_iter++;

  }

  // Then try creating a store if we have a default store defined

  if (store_list == NULL) {

    if (defaultStore != NULL) {

      createCategoryFromModel(category, defaultStore);

      category_map_t::iterator cat_iter = pcategories->find(category);

      if (cat_iter != pcategories->end()) {

        store_list = cat_iter->second;

      } else {

        LOG_OPER("failed to create new default store for category <%s>",

                 category.c_str());

      }

    }

  }

  return store_list;

}

// Add this message to every store in list

void scribeHandler::addMessage(

  const LogEntry& entry,

  const shared_ptr<store_list_t>& store_list) {

  int numstores = 0;

  // Add message to store_list

  for (store_list_t::iterator store_iter = store_list->begin();

       store_iter != store_list->end();

       ++store_iter) {

    ++numstores;

    boost::shared_ptr<LogEntry> ptr(new LogEntry);

    ptr->category = entry.category;

    ptr->message = entry.message;

    (*store_iter)->addMessage(ptr);

  }

  if (numstores) {

    incrementCounter("received good");

  } else {

    incrementCounter("received bad");

  }

}

ResultCode scribeHandler::Log(const vector<LogEntry>&  messages) {

  ResultCode result;

  scribeHandlerLock.acquireRead();

  if (throttleRequest(messages)) {

    result = TRY_LATER;

    goto end;

  }

  for (vector<LogEntry>::const_iterator msg_iter = messages.begin();

       msg_iter != messages.end();

       ++msg_iter) {

    // disallow blank category from the start

    if ((*msg_iter).category.empty()) {

      incrementCounter("received blank category");

      continue;

    }

    shared_ptr<store_list_t> store_list;

    string category = (*msg_iter).category;

    // First look for an exact match of the category

    if (pcategories) {

      category_map_t::iterator cat_iter = pcategories->find(category);

      if (cat_iter != pcategories->end()) {

        store_list = cat_iter->second;

      }

    }

    // Try creating a new store for this category if we didn't find one

    if (store_list == NULL) {

      // Need write lock to create a new category

      scribeHandlerLock.release();

      scribeHandlerLock.acquireWrite();

      store_list = createNewCategory(category);

      scribeHandlerLock.release();

      scribeHandlerLock.acquireRead();

    }

    if (store_list == NULL) {

       LOG_OPER("log entry has invalid category <%s>",

                (*msg_iter).category.c_str());

      incrementCounter("received bad");

      continue;

    }

    // Log this message

    addMessage(*msg_iter, store_list);

  }

  result = OK;

 end:

  scribeHandlerLock.release();

  return result;

}

// Returns true if overloaded.

// Allows a fixed number of messages per second.

bool scribeHandler::throttleDeny(int num_messages) {

  time_t now;

  time(&now);

  if (now != lastMsgTime) {

    lastMsgTime = now;

    numMsgLastSecond = 0;

  }

  // If we get a single huge packet it's not cool, but we'd better

  // accept it or we'll keep having to read it and deny it indefinitely

  if (num_messages > (int)maxMsgPerSecond/2) {

    LOG_OPER("throttle allowing rediculously large packet with <%d> messages", num_messages);

    return false;

  }

  if (numMsgLastSecond + num_messages > maxMsgPerSecond) {

    LOG_OPER("throttle denying request with <%d> messages. It would exceed max of <%lu> messages this second",

           num_messages, maxMsgPerSecond);

    return true;

  } else {

    numMsgLastSecond += num_messages;

    return false;

  }

}

void scribeHandler::stopStores() {

  setStatus(STOPPING);

  // Thrift doesn't currently support stopping the server from the handler,

  // so this could leave clients in weird states.

  deleteCategoryMap(pcategories);

  pcategories = NULL;

  if (pcategory_prefixes) {

    delete pcategory_prefixes;

    pcategory_prefixes = NULL;

  }

}

void scribeHandler::shutdown() {

  RWGuard monitor(scribeHandlerLock, true);

  stopStores();

  exit(0);

}

void scribeHandler::reinitialize() {

  RWGuard monitor(scribeHandlerLock, true);

  // reinitialize() will re-read the config file and re-configure the stores.

  // This is done without shutting down the Thrift server, so this will not

  // reconfigure any server settings such as port number.

  LOG_OPER("reinitializing");

  stopStores();

  initialize();

}

void scribeHandler::initialize() {

  // This clears out the error state, grep for setStatus below for details

  setStatus(STARTING);

  setStatusDetails("configuring");

  bool perfect_config = true;

  bool enough_config_to_run = true;

  int numstores = 0;

  pnew_categories = new category_map_t;

  pnew_category_prefixes = new category_prefix_map_t;

  tmpDefault.reset();

  try {

    // Get the config data and parse it.

    // If a file has been explicitly specified we'll take the conf from there,

    // which is very handy for testing and one-off applications.

    // Otherwise we'll try to get it from the service management console and

    // fall back to a default file location. This is for production.

    StoreConf config;

    string config_file;

    if (configFilename.empty()) {

      config_file = DEFAULT_CONF_FILE_LOCATION;

    } else {

      config_file = configFilename;

    }

    config.parseConfig(config_file);

    // load the global config

    config.getUnsigned("max_msg_per_second", maxMsgPerSecond);

    config.getUnsigned("max_queue_size", maxQueueSize);

    config.getUnsigned("check_interval", checkPeriod);

    // If new_thread_per_category, then we will create a new thread/StoreQueue

    // for every unique message category seen.  Otherwise, we will just create

    // one thread for each top-level store defined in the config file.

    string temp;

    config.getString("new_thread_per_category", temp);

    if (0 == temp.compare("no")) {

      newThreadPerCategory = false;

    } else {

      newThreadPerCategory = true;

    }

    unsigned long int old_port = port;

    config.getUnsigned("port", port);

    if (old_port != 0 && port != old_port) {

      LOG_OPER("port %lu from conf file overriding old port %lu", port, old_port);

    }

    if (port <= 0) {

      throw runtime_error("No port number configured");

    }

    // check if config sets the size to use for the ThreadManager

    unsigned long int num_threads;

    if (config.getUnsigned("num_thrift_server_threads", num_threads)) {

      numThriftServerThreads = (size_t) num_threads;

      if (numThriftServerThreads <= 0) {

        LOG_OPER("invalid value for num_thrift_server_threads: %lu",

                 num_threads);

        throw runtime_error("invalid value for num_thrift_server_threads");

      }

    }

    // Build a new map of stores, and move stores from the old map as

    // we find them in the config file. Any stores left in the old map

    // at the end will be deleted.

    std::vector<pStoreConf> store_confs;

    config.getAllStores(store_confs);

    for (std::vector<pStoreConf>::iterator iter = store_confs.begin();

         iter != store_confs.end();

         ++iter) {

        pStoreConf store_conf = (*iter);

        bool success = configureStore(store_conf, &numstores);

        if (!success) {

          perfect_config = false;

        }

    }

  } catch(std::exception const& e) {

    string errormsg("Bad config - exception: ");

    errormsg += e.what();

    setStatusDetails(errormsg);

    perfect_config = false;

    enough_config_to_run = false;

  }

  if (numstores) {

    LOG_OPER("configured <%d> stores", numstores);

  } else {

    setStatusDetails("No stores configured successfully");

    perfect_config = false;

    enough_config_to_run = false;

  }

  // clean up existing stores

  deleteCategoryMap(pcategories);

  pcategories = NULL;

  if (pcategory_prefixes) {

    delete pcategory_prefixes;

    pcategory_prefixes = NULL;

  }

  defaultStore.reset();

  if (enough_config_to_run) {

    pcategories = pnew_categories;

    pcategory_prefixes = pnew_category_prefixes;

    defaultStore = tmpDefault;

  } else {

    // If the new configuration failed we'll run with

    // nothing configured and status set to WARNING

    deleteCategoryMap(pnew_categories);

    if (pnew_category_prefixes) {

      delete pnew_category_prefixes;

    }

  }

  pnew_categories = NULL;

  pnew_category_prefixes = NULL;

  tmpDefault.reset();

  if (!perfect_config || !enough_config_to_run) { // perfect should be a subset of enough, but just in case

    setStatus(WARNING); // status details should have been set above

  } else {

    setStatusDetails("");

    setStatus(ALIVE);

  }

}

// Configures the store specified by the store configuration. Returns false if failed.

bool scribeHandler::configureStore(pStoreConf store_conf, int *numstores) {

  string category;

  shared_ptr<StoreQueue> pstore;

  vector<string> category_list;

  shared_ptr<StoreQueue> model;

  bool single_category = true;

  // Check if a single category is specified

  if (store_conf->getString("category", category)) {

    category_list.push_back(category);

  }

  // Check if multiple categories are specified

  string categories;

  if (store_conf->getString("categories", categories)) {

    // We want to set up to configure multiple categories, even if there is

    // only one category specified here so that configuration is consistent

    // for the 'categories' keyword.

    single_category = false;

    // Parse category names, separated by whitespace

    stringstream ss(categories);

    while (ss >> category) {

      category_list.push_back(category);

    }

  }

  if (category_list.size() == 0) {

    setStatusDetails("Bad config - store with no category");

    return false;

  }

  else if (single_category) {

    // configure single store

    shared_ptr<StoreQueue> result =

      configureStoreCategory(store_conf, category_list[0], model);

    if (result == NULL) {

      return false;

    }

    (*numstores)++;

  } else {

    // configure multiple stores

    string type;

    if (!store_conf->getString("type", type) ||

        type.empty()) {

      string errormsg("Bad config - no type for store with category: ");

      errormsg += categories;

      setStatusDetails(errormsg);

      return false;

    }

    // create model so that we can create stores as copies of this model

    model = configureStoreCategory(store_conf, categories, model, true);

    if (model == NULL) {

      string errormsg("Bad config - could not create store for category: ");

      errormsg += categories;

      setStatusDetails(errormsg);

      return false;

    }

    // create a store for each category

    vector<string>::iterator iter;

    for (iter = category_list.begin(); iter < category_list.end(); iter++) {

       shared_ptr<StoreQueue> result =

         configureStoreCategory(store_conf, *iter, model);

      if (!result) {

        return false;

      }

      (*numstores)++;

    }

  }

  return true;

}

// Configures the store specified by the store configuration and category.

shared_ptr<StoreQueue> scribeHandler::configureStoreCategory(

  pStoreConf store_conf,                       //configuration for store

  const string &category,                      //category name

  const boost::shared_ptr<StoreQueue> &model,  //model to use (optional)

  bool category_list) {                        //is a list of stores?

  bool is_default = false;

  bool already_created = false;

  if (category.empty()) {

    setStatusDetails("Bad config - store with blank category");

    return shared_ptr<StoreQueue>();

  }

  LOG_OPER("CATEGORY : %s", category.c_str());

  if (0 == category.compare("default")) {

    if (tmpDefault != NULL) {

      setStatusDetails("Bad config - multiple default stores specified");

      return shared_ptr<StoreQueue>();

    }

    is_default = true;

  }

  bool is_prefix_category = (!category.empty() &&

                             category[category.size() - 1] == '*' &&

                             !category_list);

  std::string type;

  if (!store_conf->getString("type", type) ||

      type.empty()) {

    string errormsg("Bad config - no type for store with category: ");

    errormsg += category;

    setStatusDetails(errormsg);

    return shared_ptr<StoreQueue>();

  }

  // look for the store in the current list

  shared_ptr<StoreQueue> pstore;

  if (!is_prefix_category && pcategories) {

    category_map_t::iterator category_iter = pcategories->find(category);

    if (category_iter != pcategories->end()) {

      shared_ptr<store_list_t> pstores = category_iter->second;

      for ( store_list_t::iterator it = pstores->begin(); it != pstores->end(); ++it ) {

        if ( (*it)->getBaseType() == type &&

             pstores->size() <=  1) { // no good way to match them up if there's more than one

          pstore = (*it);

          pstores->erase(it);

        }

      }

    }

  }

  try {

    // create a new store if it doesn't already exist

    if (!pstore) {

      if (model != NULL) {

        // Create a copy of the model if we want a new thread per category

        if (newThreadPerCategory && !is_default && !is_prefix_category) {

          pstore = shared_ptr<StoreQueue>(new StoreQueue(model, category));

        } else {

          pstore = model;

          already_created = true;

        }

      } else {

        string store_name;

        bool is_model, multi_category, categories;

        /* remove any *'s from category name */

        if (is_prefix_category)

          store_name = category.substr(0, category.size() - 1);

        else

          store_name = category;

        // Does this store define multiple categories

        categories = (is_default || is_prefix_category || category_list);

        // Determine if this store will actually handle multiple categories

        multi_category = !newThreadPerCategory && categories;

        // Determine if this store is just a model for later stores

        is_model = newThreadPerCategory && categories;

        pstore =

          shared_ptr<StoreQueue>(new StoreQueue(type, store_name, checkPeriod,

                                                is_model, multi_category));

      }

    }

  } catch (...) {

    pstore.reset();

  }

  if (!pstore) {

    string errormsg("Bad config - can't create a store of type: ");

    errormsg += type;

    setStatusDetails(errormsg);

    return shared_ptr<StoreQueue>();

  }

  // open store. and configure it if not copied from a model

  if (model == NULL) {

    pstore->configureAndOpen(store_conf);

  } else if (!already_created) {

    pstore->open();

  }

  if (is_default) {

    LOG_OPER("Creating default store");

    tmpDefault = pstore;

  }

  else if (is_prefix_category) {

    category_prefix_map_t::iterator category_iter =

      pnew_category_prefixes->find(category);

    if (category_iter == pnew_category_prefixes->end()) {

      (*pnew_category_prefixes)[category] = pstore;

    } else {

      string errormsg =

        "Bad config - multiple prefix stores specified for category: ";

      errormsg += category;

      setStatusDetails(errormsg);

      return shared_ptr<StoreQueue>();

    }

  }

  // push the new store onto the new map if it's not just a model

  if (!pstore->isModelStore() && !category_list) {

    shared_ptr<store_list_t> pstores;

    category_map_t::iterator category_iter = pnew_categories->find(category);

    if (category_iter != pnew_categories->end()) {

      pstores = category_iter->second;

    } else {

      pstores = shared_ptr<store_list_t>(new store_list_t);

      (*pnew_categories)[category] = pstores;

    }

        pstores->push_back(pstore);

  }

  return pstore;

}

// delete pcats and everything it contains

void scribeHandler::deleteCategoryMap(category_map_t *pcats) {

  if (!pcats) {

    return;

  }

  for (category_map_t::iterator cat_iter = pcats->begin();

       cat_iter != pcats->end();

       ++cat_iter) {

    shared_ptr<store_list_t> pstores = cat_iter->second;

    if (!pstores) {

      throw std::logic_error("deleteCategoryMap: iterator in category map holds null pointer");

    }

    for (store_list_t::iterator store_iter = pstores->begin();

         store_iter != pstores->end();

         ++store_iter) {

      if (!*store_iter) {

        throw std::logic_error("deleteCategoryMap: iterator in store map holds null pointer");

      }

      (*store_iter)->stop();

    } // for each store

    pstores->clear();

  } // for each category

  pcats->clear();

  delete pcats;

}