text:
/* LICENSE:
Copyright (c) Members of the EGEE Collaboration. 2010.
See http://www.eu-egee.org/partners/ for details on the copyright
holders.

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.

END LICENSE */
#include <cstdlib>
#include <cstdio>
#include <cerrno>

#include <iostream>
#include <string>

#include <classad_distribution.h>

#include <boost/filesystem/path.hpp>
#include <boost/filesystem/operations.hpp>

namespace fs = boost::filesystem;

#include <openssl/pem.h>
#include <openssl/x509.h>

#include "common/src/utilities/boost_fs_add.h"
#include "glite/lb/producer.h"
#include "glite/lb/consumer.h"
#include "glite/lb/context.h"
#include "glite/wms/common/configuration/Configuration.h"
#include "common/src/configuration/CommonConfiguration.h"
#include "glite/wms/common/logger/manipulators.h"
#include "glite/wms/common/logger/edglog.h"

#include "jobcontrol_namespace.h"

#include "SignalChecker.h"
#include "EventLogger.h"

using namespace std;
USING_COMMON_NAMESPACE;
RenameLogStreamNS_ts( ts );
RenameLogStreamNS( elog );

JOBCONTROL_NAMESPACE_BEGIN {

namespace jccommon {

namespace {

  // retrieve the subject_name from a given x509_proxy (thx to giaco)
  std::string get_proxy_subject(std::string const& x509_proxy)
  {
    static std::string const null_string;

    std::FILE* fd = std::fopen(x509_proxy.c_str(), "r");
    if (!fd) return null_string;
    boost::shared_ptr<std::FILE> fd_(fd, std::fclose);

    ::X509* const cert = ::PEM_read_X509(fd, 0, 0, 0);
    if (!cert) return null_string;
    boost::shared_ptr< ::X509> cert_(cert, ::X509_free);

    char* const s = ::X509_NAME_oneline(::X509_get_subject_name(cert), 0, 0);
    if (!s) return null_string;
    boost::shared_ptr<char> s_(s, ::free);

    return std::string(s);
  }

}

unsigned int  EventLogger::el_s_retries = 3, EventLogger::el_s_sleep = 30;
const char   *EventLogger::el_s_notLogged = "Event not logged, context unset.", *EventLogger::el_s_unavailable = "unavailable";
const char   *EventLogger::el_s_OK = "OK", *EventLogger::el_s_failed = "Failed";

LoggerException::LoggerException( const char *reason ) : le_reason( reason ? reason : "" )
{}

LoggerException::LoggerException( const string &reason ) : le_reason( reason )
{}

LoggerException::~LoggerException( void ) throw() {}

const char *LoggerException::what( void ) const throw()
{
  return this->le_reason.c_str();
}

string EventLogger::getLoggingError( const char *preamble )
{
  string       cause( preamble ? preamble : "" );

  if( preamble ) cause.append( 1, ' ' );

  char        *text, *desc;

  edg_wll_Error( *this->el_context, &text, &desc );
  cause.append( text );
  cause.append( " - " ); cause.append( desc );

  free( desc ); free( text );

  return cause;
}

void EventLogger::test_lb_exitcode( int &code, bool retry )
{
  const configuration::CommonConfiguration     *conf = configuration::Configuration::instance()->common();
  int          ret;
  string       cause, host_proxy;

  if( code ) {
    cause = this->getLoggingError( NULL );

    switch( code ) {
    case EINVAL:
      ts::edglog << logger::setlevel( logger::critical )
		 << "Critical error in L&B calls: EINVAL." << endl
		 << "Cause = \"" << cause << "\"." << endl;

      code = 0; // Don't retry...
      break;
    case 2:
    case EDG_WLL_ERROR_GSS:
      if (EDG_WLL_ERROR_GSS == code) {
        ts::edglog << logger::setlevel( logger::severe )
        << "Severe error in GSS layer while communicating with L&B daemons." << endl
        << "Cause = \"" << cause << "\"." << endl;
      }

      if( this->el_hostProxy ) {
         ts::edglog << "The log with the host certificate has just been done. Giving up." << endl;
         code = 0; // Don't retry...
      } else {
        ts::edglog << logger::setlevel( logger::info )
		    << "Retrying using host proxy certificate..." << endl;
        host_proxy = conf->host_proxy_file();
       if( host_proxy.length() == 0 ) {
         ts::edglog << logger::setlevel( logger::warning )
           << "Host proxy file not set inside configuration file." << endl
           << "Trying with a default NULL and hoping for the best." << endl;

       ret = edg_wll_SetParam( *this->el_context, EDG_WLL_PARAM_X509_PROXY, NULL );
       }	else {
          ts::edglog << logger::setlevel( logger::info )
		      << "Host proxy file found = \"" << host_proxy << "\"." << endl;
          ret = edg_wll_SetParam( *this->el_context, EDG_WLL_PARAM_X509_PROXY, host_proxy.c_str() );
       }
       if( ret ) {
          ts::edglog << logger::setlevel( logger::severe )
		       << "Cannot set the host proxy inside the context. Giving up." << endl;
       code = 0; // Don't retry.
	} else {
      this->el_hostProxy = true; // Set and retry (code is still != 0)
   }
   }

      break;
    default:
      if( ++this->el_count > el_s_retries ) {
	ts::edglog << logger::setlevel( logger::error )
		   << "L&B call retried " << this->el_count << " times always failed." << endl
		   << "Ignoring." << endl;

	code = 0; // Don't retry anymore
      }
      else {
	ts::edglog << logger::setlevel( logger::warning )
		   << "L&B call got an error (" << code << "). Waiting " << el_s_sleep << " seconds and trying again." << endl
		   << logger::setlevel( logger::info )
		   << "Try n. " << this->el_count << "/" << el_s_retries << endl;

	sleep( el_s_sleep );
      }
      break;
    }
  }
  else // The logging call worked fine, do nothing
    ts::edglog << logger::setlevel( logger::debug ) << "L&B call succeeded." << endl;

  SignalChecker::instance()->throw_on_signal();

  return;
}

EventLogger::EventLogger( void ) : el_remove( true ),
				   el_flag( EDG_WLL_SEQ_NORMAL ),
				   el_count( 0 ), el_context( NULL ), el_proxy(), el_have_lbproxy(true)
{
  this->el_context = new edg_wll_Context;

  if( edg_wll_InitContext(this->el_context) )
    throw LoggerException( "Cannot initialize logging context" );
  const configuration::CommonConfiguration     *common = configuration::Configuration::instance()->common();
  this->el_have_lbproxy = common->lbproxy();
}

EventLogger::EventLogger( edg_wll_Context *cont, int flag ) : el_remove( false ), el_hostProxy( false ),
							      el_flag( flag ), el_count( 0 ), el_context( cont ),
							      el_proxy(), el_have_lbproxy(true)
{
  const configuration::CommonConfiguration *common
    = configuration::Configuration::instance()->common();
  this->el_have_lbproxy = common->lbproxy();
}

EventLogger::~EventLogger( void )
{
  if( this->el_context && this->el_remove ) {
    edg_wll_FreeContext( *this->el_context );

    delete this->el_context;
  }
}

EventLogger &EventLogger::initialize_jobcontroller_context( ProxySet *ps )
{
  int   res = 0;

  if( this->el_context ) {
    res = edg_wll_SetParam( *this->el_context, EDG_WLL_PARAM_SOURCE, EDG_WLL_SOURCE_JOB_SUBMISSION );
    res |= edg_wll_SetParam( *this->el_context, EDG_WLL_PARAM_INSTANCE, "unique" );

    if( ps ) {
      res |= edg_wll_SetParam( *this->el_context, EDG_WLL_PARAM_X509_PROXY, ps->ps_x509Proxy );
      res |= edg_wll_SetParam( *this->el_context, EDG_WLL_PARAM_X509_KEY, ps->ps_x509Key );
      res |= edg_wll_SetParam( *this->el_context, EDG_WLL_PARAM_X509_CERT, ps->ps_x509Cert );
    }

    if( res ) throw LoggerException( "Invalid context parameter setting." );
  }

  return *this;
}

EventLogger &EventLogger::initialize_logmonitor_context( ProxySet *ps )
{
  int    res = 0;

  if( this->el_context ) {
    res = edg_wll_SetParam( *this->el_context, EDG_WLL_PARAM_SOURCE, EDG_WLL_SOURCE_LOG_MONITOR );
    res |= edg_wll_SetParam( *this->el_context, EDG_WLL_PARAM_INSTANCE, "unique" );

    if( ps ) {
      res |= edg_wll_SetParam( *this->el_context, EDG_WLL_PARAM_X509_PROXY, ps->ps_x509Proxy );
      res |= edg_wll_SetParam( *this->el_context, EDG_WLL_PARAM_X509_KEY, ps->ps_x509Key );
      res |= edg_wll_SetParam( *this->el_context, EDG_WLL_PARAM_X509_CERT, ps->ps_x509Cert );
    }

    if( res ) throw LoggerException( "Invalid context parameter setting." );
  }

  return *this;
}

EventLogger &EventLogger::reset_context( const string &jobid, const string &sequence, int flag )
{
  int             res;
  edg_wlc_JobId   id;

  if( this->el_context ) {
    edg_wlc_JobIdParse( jobid.c_str(), &id );
    res = edg_wll_SetLoggingJob( *this->el_context, id, sequence.c_str(), flag );
    edg_wlc_JobIdFree( id );
    if( res != 0 ) throw LoggerException( this->getLoggingError("Cannot reset logging context:") );
  }

  return *this;
}

EventLogger &EventLogger::reset_user_proxy( const string &proxyfile )
{
  bool    erase = false;
  int     res;

  if( proxyfile.size() && (proxyfile != this->el_proxy) ) {
    fs::path    pf(utilities::normalize_path(proxyfile), fs::native);

    if( fs::exists(pf) ) {
      this->el_proxy.assign( proxyfile );

      res = edg_wll_SetParam( *this->el_context, EDG_WLL_PARAM_X509_PROXY, proxyfile.c_str() );

      if( res ) throw LoggerException( this->getLoggingError("Cannot set proxyfile path inside context:") );
    }
    else erase = true;
  }
  else if( proxyfile.size() == 0 ) erase = true;

  if( erase ) {
    this->el_proxy.erase();

    res = edg_wll_SetParam( *this->el_context, EDG_WLL_PARAM_X509_PROXY, NULL );

    if( res ) throw LoggerException( this->getLoggingError("Cannot reset proxyfile path inside context:") );
  }

  return *this;
}

EventLogger &EventLogger::reset_context( const string &jobid, const string &sequence )
{
  int             res;
  edg_wlc_JobId   id;

  if( this->el_context ) {
    edg_wlc_JobIdParse( jobid.c_str(), &id );
    res = edg_wll_SetLoggingJob( *this->el_context, id, sequence.c_str(), this->el_flag );
    edg_wlc_JobIdFree( id );
    if( res != 0 ) throw LoggerException( this->getLoggingError("Cannot reset logging context:") );
  }

  return *this;
}

EventLogger &EventLogger::set_LBProxy_context( const string &jobid, const string &sequence, const string &proxyfile)
{
  bool              erase = false;
  int               res;
  edg_wlc_JobId     id;

  if( proxyfile.size() && (proxyfile != this->el_proxy) ) {
    fs::path    pf( utilities::normalize_path(proxyfile), fs::native );
    if( fs::exists(pf) ) {
      this->el_proxy.assign( proxyfile );

      res = edg_wll_SetParam( *this->el_context, EDG_WLL_PARAM_X509_PROXY, proxyfile.c_str() );

      if( res ) throw LoggerException( this->getLoggingError("Cannot set proxyfile path inside context:") );
    }
    else erase = true;
  }
  else if( proxyfile.size() == 0 ) erase = true;

  if( erase ) {
    this->el_proxy.erase();

    res = edg_wll_SetParam( *this->el_context, EDG_WLL_PARAM_X509_PROXY, NULL );
    if( res ) throw LoggerException( this->getLoggingError("Cannot reset proxyfile path inside context:") );
  }

  if( this->el_context ) {
    std::string const user_dn = get_proxy_subject( proxyfile );
    edg_wlc_JobIdParse( jobid.c_str(), &id );
    res = edg_wll_SetLoggingJobProxy( *this->el_context, id, sequence.c_str(), user_dn.c_str(), this->el_flag );
    edg_wlc_JobIdFree( id );
    if( res != 0 ) throw LoggerException( this->getLoggingError("Cannot set LBProxy context:") );
  }

  return *this;
}

void EventLogger::unhandled_event( const char *descr )
{
  logger::StatePusher     pusher( elog::cedglog, "EventLogger::unhandled_event(...)" );

  elog::cedglog << logger::setlevel( logger::ugly )
		<< "Unhandled event, what to do ?" << endl
		<< "Logging nothing..." << endl;

  return;
}

void EventLogger::condor_submit_event( const string &condorId, const string &rsl )
{
  logger::StatePusher     pusher( elog::cedglog, "EventLogger::condor_submit_event(...)" );

  int          res;

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogAcceptedProxy( *this->el_context, EDG_WLL_SOURCE_JOB_SUBMISSION, "localhost", el_s_unavailable, condorId.c_str() );
      } else {
        res = edg_wll_LogAccepted( *this->el_context, EDG_WLL_SOURCE_JOB_SUBMISSION, "localhost", el_s_unavailable, condorId.c_str() );
      }
      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else {
    elog::cedglog << logger::setlevel( logger::null )
		  << "Got condor submit event, condor id = " << condorId << endl
		  << el_s_notLogged << endl;
  }

  return;
}

void EventLogger::globus_submit_event( const string &ce, const string &rsl, const string &logfile )
{
  logger::StatePusher     pusher( elog::cedglog, "EventLogger::globus_submit_event(...)" );

  int           res;

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogTransferOKProxy( *this->el_context, EDG_WLL_SOURCE_LRMS, ce.c_str(),
                                        logfile.c_str(), rsl.c_str(), "Job successfully submitted to Globus", el_s_unavailable );
      } else {
        res = edg_wll_LogTransferOK( *this->el_context, EDG_WLL_SOURCE_LRMS, ce.c_str(),
				   logfile.c_str(), rsl.c_str(), "Job successfully submitted to Globus",
				   el_s_unavailable );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  } else {
    elog::cedglog << logger::setlevel( logger::null )
		  << "Got globus submit event, ce = " << ce << endl
		  << el_s_notLogged << endl;
  }

  return;
}

void EventLogger::grid_submit_event( const string &ce, const string &logfile )
{
  logger::StatePusher     pusher( elog::cedglog, "EventLogger::grid_submit_event(...)" );

  int           res;

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogTransferOKProxy( *this->el_context, EDG_WLL_SOURCE_LRMS, ce.c_str(),
                                   logfile.c_str(), "Grid job - no RSL", "Job successfully submitted over the Grid",
                                   el_s_unavailable );
      } else {
        res = edg_wll_LogTransferOK( *this->el_context, EDG_WLL_SOURCE_LRMS, ce.c_str(),
				   logfile.c_str(), "Grid job - no RSL", "Job successfully submitted over the Grid",
				   el_s_unavailable );
      }
      this->test_lb_exitcode( res );
    } while( res != 0 );
  } else {
    elog::cedglog << logger::setlevel( logger::null )
		  << "Got grid submit event, ce = " << ce << endl
		  << el_s_notLogged << endl;
  }

  return;
}

void EventLogger::execute_event( const char *host )
{
  logger::StatePusher     pusher( elog::cedglog, "EventLogger::execute_event(...)" );

  int           res;

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogRunningProxy( *this->el_context, host );
      } else {
        res = edg_wll_LogRunning( *this->el_context, host );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  } else {
    elog::cedglog << logger::setlevel( logger::null )
		  << "Got job execute event, host = " << host << endl
		  << el_s_notLogged << endl;
  }

  return;
}

void EventLogger::terminated_event(int retcode)
{
  logger::StatePusher     pusher( elog::cedglog, "EventLogger::terminated_event(...)" );

  int          res;
  std::string reason_to_log;

  if (retcode) {
    reason_to_log = "Warning: job exit code != 0";
  } else {
    reason_to_log = "Job terminated successfully";
  }

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogDoneOKProxy(*this->el_context, reason_to_log.c_str(), retcode);
      } else {
        res = edg_wll_LogDoneOK(*this->el_context, reason_to_log.c_str(), retcode);
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  } else {
    elog::cedglog << logger::setlevel( logger::null )
		  << "Got job terminated event, return code = " << retcode << endl
		  << el_s_notLogged << endl;
  }

  return;
}

void EventLogger::failed_on_error_event(const string &cause)
{
  logger::StatePusher      pusher( elog::cedglog, "EventLogger::failed_on_error_event(...)" );

  int           res;

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogDoneFAILEDProxy( *this->el_context, cause.c_str(), 1 );
      } else {
        res = edg_wll_LogDoneFAILED( *this->el_context, cause.c_str(), 1 );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  } else {
    elog::cedglog << logger::setlevel( logger::null )
		  << "Got a job failed event." << endl
		  << "Reason = \"" << cause << "\"." << endl
		  << el_s_notLogged << endl;
  }

  return;
}

void EventLogger::abort_on_error_event( const string &cause )
{
  logger::StatePusher      pusher( elog::cedglog, "EventLogger::abort_on_error_event(...)" );

  int           res;

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogAbortProxy( *this->el_context, cause.c_str() );
      } else {
        res = edg_wll_LogAbort( *this->el_context, cause.c_str() );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  } else {
    elog::cedglog << logger::setlevel( logger::null )
		  << "Got a job aborted event." << endl
		  << "Reason = \"" << cause << "\"." << endl
		  << el_s_notLogged << endl;
  }

  return;
}

void EventLogger::aborted_by_system_event( const string &cause )
{
  logger::StatePusher        pusher( elog::cedglog, "EventLogger::abort_by_system_event(...)" );

  int           res;

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogDoneFAILEDProxy( *this->el_context, cause.c_str(), 1 );
      } else {
        res = edg_wll_LogDoneFAILED( *this->el_context, cause.c_str(), 1 );
      }
      this->test_lb_exitcode( res );
    } while( res != 0 );
  } else {
    elog::cedglog << logger::setlevel( logger::null )
		  << "Got aborted by system event." << endl
		  << "Cause = \"" << cause << "\"" << endl
		  << el_s_notLogged << endl;
  }

  return;
}

void EventLogger::aborted_by_user_event( void )
{
  logger::StatePusher       pusher( elog::cedglog, "EventLogger::aborted_by_user_event()" );

  int           res;

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogCancelDONEProxy( *this->el_context, "Aborted by user." );
      } else {
        res = edg_wll_LogCancelDONE( *this->el_context, "Aborted by user." );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );

    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogDoneCANCELLEDProxy( *this->el_context, "Aborted by user", 0 );
      } else {
        res = edg_wll_LogDoneCANCELLED( *this->el_context, "Aborted by user", 0 );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  } else {
    elog::cedglog << logger::setlevel( logger::null )
		  << "Got aborted by user event." << endl
		  << el_s_notLogged << endl;
  }

  return;
}

void EventLogger::globus_submit_failed_event( const string &rsl, const char *reason, const string &logfile )
{
  logger::StatePusher      pusher( elog::cedglog, "EventLogger::globus_submit_failed_event(...)" );

  int           res;

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogTransferFAILProxy( *this->el_context, EDG_WLL_SOURCE_LRMS, el_s_unavailable, logfile.c_str(),
					  rsl.c_str(), reason, el_s_unavailable );
      } else {
        res = edg_wll_LogTransferFAIL( *this->el_context, EDG_WLL_SOURCE_LRMS, el_s_unavailable, logfile.c_str(),
				     rsl.c_str(), reason, el_s_unavailable );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  } else {
    elog::cedglog << logger::setlevel( logger::null )
		  << "Got globus submission failed event." << endl
		  << "Reason = \"" << reason << "\"" << endl
		  << el_s_notLogged << endl;
  }

  return;
}

/*
void EventLogger::globus_resource_down_event( void )
{
  logger::StatePusher      pusher( elog::cedglog, "EventLogger::globus_resource_down_event()" );

  int           res;

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogDoneFAILEDProxy( *this->el_context, "Globus resource down", 1 );
      } else {
        res = edg_wll_LogDoneFAILED( *this->el_context, "Globus resource down", 1 );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else
    elog::cedglog << logger::setlevel( logger::null )
		  << "Got Globus resource down event." << endl
		  << el_s_notLogged << endl;

  return;
}
*/
void EventLogger::job_held_event( const string &reason )
{
  logger::StatePusher     pusher( elog::cedglog, "EventLogger::job_held_event(...)" );

  int            res;
  string         what( "Got a job held event, reason: " );

  what.append( reason );

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogDoneFAILEDProxy( *this->el_context, what.c_str(), 1 );
      } else {
        res = edg_wll_LogDoneFAILED( *this->el_context, what.c_str(), 1 );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else
    elog::cedglog << logger::setlevel( logger::null )
		  << "Got job held event." << endl
		  << "Reason = \"" << reason << "\"" << endl
		  << el_s_notLogged << endl;
}

void EventLogger::job_enqueued_start_event( const string &filename, const classad::ClassAd *ad )
{
  logger::StatePusher      pusher( ts::edglog, "EventLogger::job_enqueued_start_event(...)" );

  int                         res;
  string                      job;
  classad::ClassAdUnParser    unparser;

  if( ad != NULL ) unparser.Unparse( job, (classad::ClassAd *) ad );

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogEnQueuedSTARTProxy( *this->el_context, filename.c_str(), job.c_str(), el_s_unavailable );
      } else {
        res = edg_wll_LogEnQueuedSTART( *this->el_context, filename.c_str(), job.c_str(), el_s_unavailable );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else
    ts::edglog << logger::setlevel( logger::null )
	       << "Job enqueued start event." << endl
	       << el_s_notLogged << endl;

  return;
}

void EventLogger::job_enqueued_ok_event( const string &filename, const classad::ClassAd *ad )
{
  logger::StatePusher      pusher( ts::edglog, "EventLogger::job_enqueued_ok_event(...)" );

  int                         res;
  string                      job;
  classad::ClassAdUnParser    unparser;

  unparser.Unparse( job, (classad::ClassAd *) ad );

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogEnQueuedOKProxy( *this->el_context, filename.c_str(), job.c_str(), el_s_unavailable );
      } else {
        res = edg_wll_LogEnQueuedOK( *this->el_context, filename.c_str(), job.c_str(), el_s_unavailable );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else
    ts::edglog << logger::setlevel( logger::null )
	       << "Job enqueued ok event." << endl
	       << el_s_notLogged << endl;

  return;
}

void EventLogger::job_enqueued_failed_event( const string &filename, const string &error, const classad::ClassAd *ad )
{
  logger::StatePusher       pusher( ts::edglog, "EventLogger::job_enqueued_failed_event(...)" );

  int                        res;
  string                     job;
  classad::ClassAdUnParser   unparser;

  unparser.Unparse( job, (classad::ClassAd *) ad );

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogEnQueuedFAILProxy( *this->el_context, filename.c_str(), job.c_str(), error.c_str() );
      } else {
        res = edg_wll_LogEnQueuedFAIL( *this->el_context, filename.c_str(), job.c_str(), error.c_str() );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else
    ts::edglog << logger::setlevel( logger::null )
	       << "Job enqueued failed." << endl
	       << "Reason = \"" << error << "\"" << endl
	       << el_s_notLogged << endl;

  return;
}

void EventLogger::job_dequeued_event( const string &filename )
{
  logger::StatePusher     pusher( elog::cedglog, "EventLogger::job_dequeued_event(...)" );

  int            res;

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogDeQueuedProxy( *this->el_context, filename.c_str(), el_s_unavailable );
      } else {
        res = edg_wll_LogDeQueued( *this->el_context, filename.c_str(), el_s_unavailable );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else
    elog::cedglog << logger::setlevel( logger::null )
		  << "Job dequeued from file " << filename << endl
		  << el_s_notLogged << endl;

  return;
}

void EventLogger::job_cancel_requested_event( const string &source )
{
  logger::StatePusher     pusher( elog::cedglog, "EventLogger::job_cancel_requested_event()" );

  int            res;
  string         reason( "Cancel requested by " );

  reason.append( source );

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogCancelREQProxy( *this->el_context, reason.c_str() );
      } else {
        res = edg_wll_LogCancelREQ( *this->el_context, reason.c_str() );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else
    elog::cedglog << logger::setlevel( logger::null )
		  << "Got cancel from " << source << endl
		  << el_s_notLogged << endl;

  return;
}

void EventLogger::condor_submit_start_event( const string &logfile )
{
  logger::StatePusher      pusher( elog::cedglog, "EventLogger::condor_submit_start_event(...)" );

  int            res;

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogTransferSTARTProxy( *this->el_context, EDG_WLL_SOURCE_LOG_MONITOR, "localhost", logfile.c_str(),
					   el_s_unavailable, el_s_unavailable, el_s_unavailable );
      } else {
        res = edg_wll_LogTransferSTART( *this->el_context, EDG_WLL_SOURCE_LOG_MONITOR, "localhost", logfile.c_str(),
				      el_s_unavailable, el_s_unavailable, el_s_unavailable );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else
    elog::cedglog << logger::setlevel( logger::null )
		  << "Condor submit start event." << endl
		  << el_s_notLogged << endl;

  return;
}

void EventLogger::condor_submit_ok_event( const string &rsl, const string &condorid, const string &logfile )
{
  logger::StatePusher     pusher( elog::cedglog, "EventLogger::condor_submit_ok_event(...)" );

  int            res;

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogTransferOKProxy( *this->el_context, EDG_WLL_SOURCE_LOG_MONITOR, "localhost", logfile.c_str(),
					rsl.c_str(), el_s_unavailable, condorid.c_str() );
      } else {
        res = edg_wll_LogTransferOK( *this->el_context, EDG_WLL_SOURCE_LOG_MONITOR, "localhost", logfile.c_str(),
				   rsl.c_str(), el_s_unavailable, condorid.c_str() );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else
    elog::cedglog << logger::setlevel( logger::null )
		  << "Condor submit ok event." << endl
		  << el_s_notLogged << endl;

  return;
}

void EventLogger::condor_submit_failed_event( const string &rsl, const string &reason, const string &logfile )
{
  logger::StatePusher      pusher( elog::cedglog, "EventLogger::condor_submit_failed_event(...)" );

  int            res;

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogTransferFAILProxy( *this->el_context, EDG_WLL_SOURCE_LOG_MONITOR, "localhost", logfile.c_str(),
					  rsl.c_str(), reason.c_str(), el_s_unavailable );
      } else {
        res = edg_wll_LogTransferFAIL( *this->el_context, EDG_WLL_SOURCE_LOG_MONITOR, "localhost", logfile.c_str(),
				     rsl.c_str(), reason.c_str(), el_s_unavailable );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );

    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogAbortProxy( *this->el_context, "Submission to condor failed." );
      } else {
        res = edg_wll_LogAbort( *this->el_context, "Submission to condor failed." );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else {
    elog::cedglog << logger::setlevel( logger::null )
		  << "Condor submit failed event." << endl
		  << logger::setmultiline( true, "CE-> " ) << "Reason\n" << reason << endl
		  << el_s_notLogged << endl;
    elog::cedglog << logger::setmultiline( false );
  }

  return;
}

void EventLogger::job_cancel_refused_event( const string &info )
{
  logger::StatePusher     pusher( elog::cedglog, "EventLogger::job_cancel_refused_event(...)" );

  int            res;

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogCancelREFUSEProxy( *this->el_context, info.c_str() );
      } else {
        res = edg_wll_LogCancelREFUSE( *this->el_context, info.c_str() );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else
    elog::cedglog << logger::setlevel( logger::null )
		  << "Cancel refused failed event." << endl
		  << "Reason \"" << info << "\"" << endl
		  << el_s_notLogged << endl;

  return;
}

void EventLogger::job_abort_classad_invalid_event( const string &logfile, const string &reason )
{
  logger::StatePusher      pusher( elog::cedglog, "EventLogger::job_abort_classad_invalid_event(...)" );

  int            res;
  string         info( "Invalid classad syntax: " );

  info.append( reason );

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogTransferFAILProxy( *this->el_context, EDG_WLL_SOURCE_LOG_MONITOR, "localhost", logfile.c_str(),
					  el_s_unavailable, info.c_str(), el_s_unavailable );
      } else {
        res = edg_wll_LogTransferFAIL( *this->el_context, EDG_WLL_SOURCE_LOG_MONITOR, "localhost", logfile.c_str(),
				     el_s_unavailable, info.c_str(), el_s_unavailable );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );

    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogAbortProxy( *this->el_context, info.c_str() );
      } else {
        res = edg_wll_LogAbort( *this->el_context, info.c_str() );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else
    elog::cedglog << logger::setlevel( logger::null )
		  << "Job aborted for invalid classad." << endl
		  << el_s_notLogged << endl;

  return;
}

void EventLogger::job_abort_cannot_write_submit_file_event( const string &logfile, const string &filename, const string &moreinfo )
{
  logger::StatePusher    pusher( elog::cedglog, "EventLogger::job_abort_cannot_write_submit_file_event(...)" );

  int            res;
  string         info( "Cannot create condor submit file \"" );

  info.append( filename ); info.append( "\": " );
  info.append( moreinfo );

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogTransferFAILProxy( *this->el_context, EDG_WLL_SOURCE_LOG_MONITOR, "localhose", logfile.c_str(),
					  el_s_unavailable, info.c_str(), el_s_unavailable );
      } else {
        res = edg_wll_LogTransferFAIL( *this->el_context, EDG_WLL_SOURCE_LOG_MONITOR, "localhose", logfile.c_str(),
				     el_s_unavailable, info.c_str(), el_s_unavailable );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );

    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogAbortProxy( *this->el_context, info.c_str() );
      } else {
        res = edg_wll_LogAbort( *this->el_context, info.c_str() );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else
    elog::cedglog << logger::setlevel( logger::null )
		  << "Job aborted for condor submit error." << endl
		  << el_s_notLogged << endl;

  return;
}

void EventLogger::job_resubmitting_event( void )
{
  logger::StatePusher     pusher( elog::cedglog, "EventLogger::job_resubmitted_event()" );

  int            res;

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
      res = edg_wll_LogResubmissionWILLRESUBProxy( *this->el_context, el_s_unavailable, el_s_unavailable );
      } else {
      res = edg_wll_LogResubmissionWILLRESUB( *this->el_context, el_s_unavailable, el_s_unavailable );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else
    elog::cedglog << logger::setlevel( logger::null )
		  << "Job resubmitting event." << endl
		  << el_s_notLogged << endl;

  return;
}

void EventLogger::job_wm_enqueued_start_event( const string &filename, const classad::ClassAd &ad )
{
  logger::StatePusher        pusher(elog::cedglog, "EventLogger::job_wm_enqueued_start_event(...)" );

  int                       res;
  string                    job;
  classad::ClassAdUnParser  unparser;

  unparser.Unparse( job, (classad::ClassAd *) &ad );

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
      res = edg_wll_LogEnQueuedSTARTProxy( *this->el_context, filename.c_str(), job.c_str(), el_s_unavailable );
      } else {
      res = edg_wll_LogEnQueuedSTART( *this->el_context, filename.c_str(), job.c_str(), el_s_unavailable );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else
    elog::cedglog << logger::setlevel( logger::null )
		  << "Job enqeueued to WM start event." << endl
		  << el_s_notLogged << endl;

  return;
}

void EventLogger::job_wm_enqueued_ok_event( const string &filename, const classad::ClassAd &ad )
{
  logger::StatePusher     pusher( elog::cedglog, "EventLogger::job_wm_enqueued_ok_event(...)" );

  int                       res;
  string                    job;
  classad::ClassAdUnParser  unparser;

  unparser.Unparse( job, (classad::ClassAd *) &ad );

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogEnQueuedOKProxy( *this->el_context, filename.c_str(), job.c_str(), el_s_unavailable );
      } else {
        res = edg_wll_LogEnQueuedOK( *this->el_context, filename.c_str(), job.c_str(), el_s_unavailable );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else
    elog::cedglog << logger::setlevel( logger::null )
		  << "Job enqueued to WM ok event." << endl
		  << el_s_notLogged << endl;

  return;
}

void EventLogger::job_wm_enqueued_failed_event( const string &filename, const classad::ClassAd &ad, const string &error )
{
  logger::StatePusher    pusher( elog::cedglog, "EventLogger::job_wm_enqueued_failed_event(...)" );

  int                       res;
  string                    job;
  classad::ClassAdUnParser  unparser;

  unparser.Unparse( job, (classad::ClassAd *) &ad );

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogEnQueuedFAILProxy( *this->el_context, filename.c_str(), job.c_str(), error.c_str() );
      } else {
        res = edg_wll_LogEnQueuedFAIL( *this->el_context, filename.c_str(), job.c_str(), error.c_str() );
      }

      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else
    elog::cedglog << logger::setlevel( logger::null )
		  << "Job enqueued to WM failed." << endl
		  << "Reason = \"" << error << "\"" << endl
		  << el_s_notLogged << endl;

  return;
}

// we need the sequence code from the WN computed by the JW.
void EventLogger::job_really_run_event( const string &sc )
{
  logger::StatePusher    pusher( elog::cedglog, "EventLogger::job_really_run_event(...)" );

  int                       res;

  if( this->el_context ) {
    this->startLogging();
    do {
      if (this->el_have_lbproxy) {
        res = edg_wll_LogReallyRunningProxy( *this->el_context, sc.c_str() );
      } else {
        res = edg_wll_LogReallyRunning( *this->el_context, sc.c_str() );
      }
      this->test_lb_exitcode( res );
    } while( res != 0 );
  }
  else
    elog::cedglog << logger::setlevel( logger::null )
                  << "Really running event." << endl
                  << el_s_notLogged << endl;

  return;
}

string EventLogger::seq_code_lbproxy( const string &jobid )
{
  char         *seqcode;
  string       res( "undefined" );
  edg_wlc_JobId       id;
  edg_wlc_JobIdParse( jobid.c_str(), &id );

      if (this->el_have_lbproxy) {
        if(this->el_context ) {
          edg_wll_QuerySequenceCodeProxy( *this->el_context, id, &seqcode );

          res.assign( seqcode );
         free( seqcode );
        }
      } else {
        res.assign( "UI=000000:NS=000000:WM=000000:BH=000000:JSS=000000:LM=000000:LRMS=000000:APP=000000" );
      }

   edg_wlc_JobIdFree( id );
  return res;

}

string EventLogger::sequence_code( void )
{
  char          *seqcode;
  string         res( "undefined" );

  if( this->el_context ) {
    seqcode = edg_wll_GetSequenceCode( *this->el_context );

    res.assign( seqcode );
    free( seqcode );
  }

  return res;
}

// Queries LB

// TO BE CHECK
string EventLogger::query_condorid( const string &jobid )
{
  string              condor_id;
  edg_wlc_JobId       id;
  edg_wlc_JobIdParse( jobid.c_str(), &id );
  edg_wll_Event       *events = NULL;
  edg_wll_QueryRec    jc[2];
  edg_wll_QueryRec    ec[3];

  memset(jc,0,sizeof jc);
  memset(ec,0,sizeof ec);

  // job condition: JOBID = jobid
  jc[0].attr    = EDG_WLL_QUERY_ATTR_JOBID;
  jc[0].op      = EDG_WLL_QUERY_OP_EQUAL;
  jc[0].value.j = id;
  jc[1].attr    = EDG_WLL_QUERY_ATTR_UNDEF;

  // event condition: Event type = ACCEPTED AND source = LM
  ec[0].attr    = EDG_WLL_QUERY_ATTR_EVENT_TYPE;
  ec[0].op      = EDG_WLL_QUERY_OP_EQUAL;
  ec[0].value.i = EDG_WLL_EVENT_ACCEPTED;
  ec[1].attr    = EDG_WLL_QUERY_ATTR_SOURCE;
  ec[1].op      = EDG_WLL_QUERY_OP_EQUAL;
  ec[1].value.i = EDG_WLL_SOURCE_LOG_MONITOR;
  ec[2].attr    = EDG_WLL_QUERY_ATTR_UNDEF;

      if (this->el_have_lbproxy) {
        edg_wll_QueryEventsProxy( *this->el_context, jc, ec, &events);
      } else {
        edg_wll_QueryEvents( *this->el_context, jc, ec, &events);
      }

  if (events) {
    for (int i = 0; events[i].type; ++i) {
      // if the job has been resubmitted we need the last one
      condor_id = events[i].accepted.local_jobid;
      edg_wll_FreeEvent(&events[i]);
    }
    free(events);
  }

  edg_wlc_JobIdFree( id );

  return condor_id;
}

} // Namespace common

} JOBCONTROL_NAMESPACE_END

max_stars_repo_name: italiangrid/wms

max_issues_repo_name: italiangrid/wms

max_forks_repo_name: italiangrid/wms


text:
#include <iostream>

#include <boost/program_options.hpp>
#include <crazyflie_cpp/Crazyradio.h>

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

  std::string addressStr;
  std::string defaultAddressStr("0xE7E7E7E7E7");

  namespace po = boost::program_options;

  po::options_description desc("Allowed options");
  desc.add_options()
    ("help", "produce help message")
    ("address", po::value<std::string>(&addressStr)->default_value(defaultAddressStr), "device address")
  ;

  try
  {
    po::variables_map vm;
    po::store(po::parse_command_line(argc, argv, desc), vm);
    po::notify(vm);

    if (vm.count("help")) {
      std::cout << desc << "\n";
      return 0;
    }
  }
  catch(po::error& e)
  {
    std::cerr << e.what() << std::endl << std::endl;
    std::cerr << desc << std::endl;
    return 1;
  }

  try
  {
    uint64_t address;
    std::stringstream sstr;
    sstr << std::hex << addressStr;
    sstr >> address;

    Crazyradio radio(0);
    radio.setAddress(address);

    for (uint8_t datarate = 0; datarate < 3; ++datarate) {
      radio.setDatarate((Crazyradio::Datarate)datarate);
      for (uint8_t channel = 0; channel <= 125; ++channel) {
        radio.setChannel(channel);

        uint8_t test[] = {0xFF};
        Crazyradio::Ack ack;
        radio.sendPacket(test, sizeof(test), ack);
        if (ack.ack) {
          std::cout << "radio://0/" << (uint32_t)channel << "/";
          switch(datarate) {
          case Crazyradio::Datarate_250KPS:
            std::cout << "250K";
            break;
          case Crazyradio::Datarate_1MPS:
            std::cout << "1M";
            break;
          case Crazyradio::Datarate_2MPS:
            std::cout << "2M";
            break;
          }

          if (defaultAddressStr != addressStr) {
            std::cout << "/" << addressStr.substr(2);
          }
          std::cout << std::endl;
        }
      }
    }
    return 0;
  }
  catch(std::exception& e)
  {
    std::cerr << e.what() << std::endl;
    return 1;
  }
}

ext: cpp

max_stars_repo_name: huisyy/autonomous-aerial-nav-hardware

max_issues_repo_name: huisyy/autonomous-aerial-nav-hardware

max_forks_repo_name: huisyy/autonomous-aerial-nav-hardware


text:
#ifndef HOOKE_LINEAR_ELASTICITY_HPP
#define HOOKE_LINEAR_ELASTICITY_HPP

#include <polyfem/Common.hpp>
#include <polyfem/ElasticityUtils.hpp>

#include <polyfem/ElementAssemblyValues.hpp>
#include <polyfem/ElementBases.hpp>

#include <polyfem/AutodiffTypes.hpp>

#include <Eigen/Dense>
#include <array>

//local assembler for HookeLinearElasticity C : (F+F^T)/2, see linear elasticity
namespace polyfem
{
	class HookeLinearElasticity
	{
	public:
		HookeLinearElasticity();

		// res is R^{dim²}
		Eigen::Matrix<double, Eigen::Dynamic, 1, 0, 9, 1>
		assemble(const ElementAssemblyValues &vals, const int i, const int j, const QuadratureVector &da) const;

		Eigen::Matrix<double, Eigen::Dynamic, 1, 0, 3, 1>
		compute_rhs(const AutodiffHessianPt &pt) const;

		void compute_von_mises_stresses(const int el_id, const ElementBases &bs, const ElementBases &gbs, const Eigen::MatrixXd &local_pts, const Eigen::MatrixXd &displacement, Eigen::MatrixXd &stresses) const;
		void compute_stress_tensor(const int el_id, const ElementBases &bs, const ElementBases &gbs, const Eigen::MatrixXd &local_pts, const Eigen::MatrixXd &displacement, Eigen::MatrixXd &tensor) const;

		inline int size() const { return size_; }

		void set_size(const int size);

		//sets the elasticty tensor
		void set_parameters(const json &params);

	private:
		int size_ = 2;

		ElasticityTensor elasticity_tensor_;

		void assign_stress_tensor(const int el_id, const ElementBases &bs, const ElementBases &gbs, const Eigen::MatrixXd &local_pts, const Eigen::MatrixXd &displacement, const int all_size, Eigen::MatrixXd &all, const std::function<Eigen::MatrixXd(const Eigen::MatrixXd &)> &fun) const;
	};
} // namespace polyfem

#endif //HOOKE_LINEAR_ELASTICITY_HPP

ext: hpp

max_stars_repo_name: danielepanozzo/polyfem

max_issues_repo_name: danielepanozzo/polyfem

max_forks_repo_name: danielepanozzo/polyfem


text:
///////////////////////////////////////////////////////////////////////////////
// statistics::survival::data::algorithm::logit_log.hpp                      //
//                                                                           //
//  Copyright 2009 Erwann Rogard. Distributed under the Boost                //
//  Software License, Version 1.0. (See accompanying file                    //
//  LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)         //
///////////////////////////////////////////////////////////////////////////////
#ifndef BOOST_STATISTICS_SURVIVAL_DATA_ALGORITHM_LOGIT_LOG_HPP_ER_2009
#define BOOST_STATISTICS_SURVIVAL_DATA_ALGORITHM_LOGIT_LOG_HPP_ER_2009
#include <stdexcept>
#include <iterator>
#include <boost/format.hpp>
#include <boost/statistics/survival/data/algorithm/detail/log_shift.hpp>

namespace boost{
namespace statistics{
namespace survival{
namespace data{

// Transforms an alternating sequence of (usually mean) failures
// and event times to their logit and log, with an offset that prevents nan.
//
// The rationale for a logit is that 0<= mean failure <=1
template<typename It,typename ItO,typename T>
ItO logit_log(
    It b,    // Alternating sequence, f[0],t[0],f[1],t[1],...
    It e,
    ItO o,   // Output iterator
    T t0,
    T t1
){
    static const char* str = "logit_log : distance(b,e) = %1% mod 2 !=2";

   if( std::distance(b,e)%2 != 0 ){
        format f(str); f%distance(b,e);
        throw std::runtime_error(f.str());
   }
   while(b!=e){
        (*o++) = detail::logit_shift((*b++),t0);
        (*o++) = detail::log_shift((*b++),t1);
    }
    return o;
}

}// data
}// survival
}// statistics
}// boost

#endif


text:
/*
// Licensed to DynamoBI Corporation (DynamoBI) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  DynamoBI licenses this file
// to you 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.
*/

#include "fennel/common/CommonPreamble.h"
#include "fennel/lcs/LcsClusterNodeWriter.h"
#include "fennel/tuple/TupleAccessor.h"
#include <boost/scoped_array.hpp>

FENNEL_BEGIN_CPPFILE("$Id$");

LcsClusterNodeWriter::LcsClusterNodeWriter(
    BTreeDescriptor const &treeDescriptorInit,
    SegmentAccessor const &accessorInit,
    TupleDescriptor const &colTupleDescInit,
    SharedTraceTarget pTraceTargetInit,
    std::string nameInit)
    : LcsClusterAccessBase(treeDescriptorInit),
      TraceSource(pTraceTargetInit, nameInit)
{
    scratchAccessor = accessorInit;
    bufferLock.accessSegment(scratchAccessor);
    bTreeWriter = SharedBTreeWriter(
        new BTreeWriter(treeDescriptorInit, scratchAccessor, true));
    colTupleDesc = colTupleDescInit;
    clusterDump =
        SharedLcsClusterDump(
            new LcsClusterDump(
                treeDescriptorInit,
                colTupleDesc,
                TRACE_FINE,
                pTraceTargetInit,
                nameInit));
    nClusterCols = 0;
    pHdr = 0;
    hdrSize = 0;
    pIndexBlock = 0;
    pBlock = 0;
    szBlock = 0;
    minSzLeft = 0;
    batchDirs.reset();
    pValBank.reset();
    oValBank.reset();
    batchOffset.reset();
    batchCount.reset();
    szLeft = 0;
    nBits.reset();
    nextWidthChange.reset();
    arraysAllocated = false;
    valBankStart.reset();
    bForceMode.reset();
    forceModeCount.reset();
    maxValueSize.reset();
}

LcsClusterNodeWriter::~LcsClusterNodeWriter()
{
    close();
}

void LcsClusterNodeWriter::close()
{
    // flush and unlock last page written
    if (clusterLock.isLocked()) {
        clusterLock.flushPage(true);
    }
    clusterLock.unlock();

    bTreeWriter.reset();
    batchDirs.reset();
    pValBank.reset();
    valBankStart.reset();
    forceModeCount.reset();
    bForceMode.reset();
    oValBank.reset();
    batchOffset.reset();
    batchCount.reset();
    nBits.reset();
    nextWidthChange.reset();
    maxValueSize.reset();
    attrAccessors.reset();
}

bool LcsClusterNodeWriter::getLastClusterPageForWrite(
    PLcsClusterNode &pBlock, LcsRid &firstRid)
{
    // get the last key in the btree (if it exists) and read the cluster
    // page based on the pageid stored in that btree record

    if (bTreeWriter->searchLast() == false) {
        bTreeWriter->endSearch();
        return false;
    }

    bTreeWriter->getTupleAccessorForRead().unmarshal(bTreeTupleData);
    clusterPageId = readClusterPageId();
    clusterLock.lockExclusive(clusterPageId);
    pBlock = &(clusterLock.getNodeForWrite());
    firstRid = pBlock->firstRID;

    // End the search so the BTreeWriter doesn't think it's positioned within
    // the btree.  We'll position properly on the first monotonic insert.
    bTreeWriter->endSearch();

    if (isTracingLevel(TRACE_FINE)) {
        FENNEL_TRACE(
            TRACE_FINE,
            "Calling ClusterDump from getLastClusterPageForWrite");
        clusterDump->dump(opaqueToInt(clusterPageId), pBlock, szBlock);
    }

    return true;
}

PLcsClusterNode LcsClusterNodeWriter::allocateClusterPage(LcsRid firstRid)
{
    // allocate a new cluster page and insert the corresponding rid, pageid
    // record into the btree

    PageId prevPageId = NULL_PAGE_ID;

    if (clusterLock.isLocked()) {
        // Remember the predecessor so that we can chain it below.
        prevPageId = clusterLock.getPageId();

        // Kick off an asynchronous write on the page we've just finished
        // so that when it comes time to checkpoint or victimize,
        // maybe it will be on disk already.
        clusterLock.flushPage(true);
    }

    clusterPageId = clusterLock.allocatePage();
    if (prevPageId != NULL_PAGE_ID) {
        segmentAccessor.pSegment->setPageSuccessor(prevPageId, clusterPageId);
    }
    bTreeRid = firstRid;
    bTreeTupleData[0].pData = reinterpret_cast<uint8_t *> (&firstRid);
    bTreeTupleData[1].pData = reinterpret_cast<uint8_t *> (&clusterPageId);
    bTreeWriter->insertTupleData(bTreeTupleData, DUP_FAIL);
    return &(clusterLock.getNodeForWrite());
}

void LcsClusterNodeWriter::init(
    uint nColumn, PBuffer iBlock, PBuffer *pB, uint szB)
{
    nClusterCols = nColumn;
    pIndexBlock = iBlock;
    pBlock = pB;
    szBlock = szB;
    pHdr = (PLcsClusterNode) pIndexBlock;

    hdrSize = getClusterSubHeaderSize(nClusterCols);

    // initialize lastVal, firstVal, and nVal fields in the header
    // to point to the appropriate positions in the indexBlock

    setHdrOffsets(pHdr);

    minSzLeft =
        nClusterCols
        * (LcsMaxLeftOver * sizeof(uint16_t) + sizeof(LcsBatchDir));

    allocArrays();
}

void LcsClusterNodeWriter::openNew(LcsRid startRID)
{
    int i;

    // Inialize block header and batches
    pHdr->firstRID = startRID;
    pHdr->nColumn = nClusterCols;
    pHdr->nBatch = 0;
    pHdr->oBatch = hdrSize;

    for (i = 0; i < nClusterCols; i++) {
        lastVal[i] = szBlock;
        firstVal[i] = (uint16_t) szBlock;
        nVal[i] = 0;
        delta[i] = 0;
        batchDirs[i].mode = LCS_COMPRESSED;
        batchDirs[i].nVal = 0;
        batchDirs[i].nRow = 0;
        batchDirs[i].oVal = 0;
        batchDirs[i].oLastValHighMark = lastVal[i];
        batchDirs[i].nValHighMark = nVal[i];
        batchOffset[i] = hdrSize;
        // # of bits it takes to represent 0 values
        nBits[i] = 0;
        nextWidthChange[i] = 1;
        batchCount[i] = 0;
    }

    // account for the header size, account for at least 1 batch for each column
    // and leave space for one addtional batch for a "left-over" batch

    szLeft =
        szBlock
        - hdrSize
        - (2 * sizeof(LcsBatchDir)) * nClusterCols;
    if (szLeft < 0) {
        szLeft = 0;
    }
}

bool LcsClusterNodeWriter::openAppend(
    uint *nValOffsets, uint16_t *lastValOffsets, RecordNum &nrows)
{
    int i;

    // leave space for one batch for each column entry
    szLeft =
        lastVal[nClusterCols - 1]
        - pHdr->oBatch
        - (pHdr->nBatch + 2* nClusterCols) * sizeof(LcsBatchDir);
    if (szLeft < 0) {
        szLeft = 0;
    }

    // Let's move the values, batch directories, and batches to
    // temporary blocks from index block
    nrows = moveFromIndexToTemp();

    for (i = 0; i < nClusterCols; i++) {
        nValOffsets[i] = nVal[i];
        lastValOffsets[i] = lastVal[i];
        memset(&batchDirs[i], 0, sizeof(LcsBatchDir));

        batchDirs[i].oLastValHighMark = lastVal[i];
        batchDirs[i].nValHighMark = nVal[i];
        batchDirs[i].mode = LCS_COMPRESSED;

        // # of bits it takes to represent 0 values
        nBits[i] = 0;
        nextWidthChange[i] = 1;

        oValBank[i] = 0;
        batchCount[i] = pHdr->nBatch / nClusterCols;
    }

    return (szLeft == 0);
}

void LcsClusterNodeWriter::describeLastBatch(
    uint column, uint &dRow, uint &recSize)
{
    PLcsBatchDir pBatch;

    pBatch = (PLcsBatchDir) (pBlock[column] + batchOffset[column]);
    dRow = pBatch[batchCount[column] -1].nRow % 8;
    recSize = pBatch[batchCount[column] -1].recSize;
}

uint16_t LcsClusterNodeWriter::getNextVal(uint column, uint16_t thisVal)
{
    if (thisVal && thisVal != szBlock) {
        return
            (uint16_t) (thisVal
                + attrAccessors[column].getStoredByteCount(
                    pBlock[column] + thisVal));
    } else {
        return 0;
    }
}

void LcsClusterNodeWriter::rollBackLastBatch(uint column, PBuffer pBuf)
{
    uint i;
    PLcsBatchDir pBatch;
    uint16_t *pValOffsets;

    uint8_t *pBit;                      // bitVecs start address
    WidthVec w;                         // bitVec width vector
    PtrVec p;                           // bitVec offsets
    uint iV;                            // # of bit vectors

    uint16_t rows[LcsMaxRollBack];      // row index storage
    int origSzLeft;
    uint len;

    // load last batch, nBatch must be at least 1
    pBatch = (PLcsBatchDir)(pBlock[column] + batchOffset[column]);
    batchDirs[column]  = pBatch[batchCount[column] -1];

    // compute size left in temporary block before roll back
    origSzLeft =
        lastVal[column]
        - batchOffset[column]
        - (batchCount[column] + 2) * sizeof(LcsBatchDir);

    if ((batchDirs[column].nRow > 8) || (batchDirs[column].nRow % 8) == 0) {
        return;
    }

    if (batchDirs[column].mode == LCS_COMPRESSED) {
        // calculate the bit vector widthes
        iV = bitVecWidth(calcWidth(batchDirs[column].nVal), w);

        // this is where the bit vectors start
        pBit = pBlock[column] + batchDirs[column].oVal
            + batchDirs[column].nVal * sizeof(uint16_t);

        // nByte are taken by the bit vectors
        bitVecPtr(batchDirs[column].nRow, iV, w, p, pBit);

        // there are at most 8 rows in this batch
        readBitVecs(rows, iV, w, p, 0, batchDirs[column].nRow);

        // get the address of the batches value offsets
        pValOffsets = (uint16_t *)(pBlock[column] + batchDirs[column].oVal);

        // fill up buffer with batches values
        for (i = 0; i < batchDirs[column].nRow;
            i++, pBuf += batchDirs[column].recSize)
        {
            len =
                attrAccessors[column].getStoredByteCount(
                    pBlock[column] + pValOffsets[rows[i]]);
            memcpy(pBuf, pBlock[column] + pValOffsets[rows[i]], len);
        }

    } else if (batchDirs[column].mode == LCS_FIXED) {
        // fixed size record batch
        // copy the values into the given buffer
        memcpy(
            pBuf,
            pBlock[column] + batchDirs[column].oVal,
            batchDirs[column].nRow * batchDirs[column].recSize);
    } else {
        // variable sized records (batch.mode == LCS_VARIABLE)
        // get the address of the batches value offsets
        pValOffsets = (uint16_t *)(pBlock[column] + batchDirs[column].oVal);

        // fill up buffer with batches values
        for (i = 0; i < batchDirs[column].nRow;
            i++, pBuf += batchDirs[column].recSize)
        {
            len =
                attrAccessors[column].getStoredByteCount(
                    pBlock[column] + pValOffsets[i]);
            memcpy(pBuf, pBlock[column] + pValOffsets[i], len);
        }
    }

    // Reset the last batch
    batchCount[column]--;
    // batch dir offset points to the beginning of the rolled back batch
    batchOffset[column] = batchDirs[column].oVal;

    // copy the batch dir back to the end of the prev batch.
    memmove(
        pBlock[column] + batchOffset[column],
        pBatch,
        batchCount[column] * sizeof(LcsBatchDir));

    // recalc size left
    // leave place for one new batch(the rolled back one will be rewriten)
    // and possibley one to follow.  Subtract the difference of the new size
    // and the original size and add this to szLeft in index variable
    int newSz;
    newSz = lastVal[column] - batchOffset[column]
        - (batchCount[column] + 2) * sizeof(LcsBatchDir);
    szLeft += (newSz - origSzLeft);
    if (szLeft < 0) {
        szLeft = 0;
    }

    // # of bits it takes to represent 0 values
    nBits[column] = 0;
    nextWidthChange[column] = 1;

    // set batch parameters
    batchDirs[column].mode = LCS_COMPRESSED;
    batchDirs[column].nVal = 0;
    batchDirs[column].nRow = 0;
    batchDirs[column].oVal = 0;
    batchDirs[column].recSize = 0;
}

// addValue() where the current value already exists

bool LcsClusterNodeWriter::addValue(uint column, bool bFirstTimeInBatch)
{
    // Calculate szleft assuming the value gets added.
    szLeft -= sizeof(uint16_t);

    // if there is not enough space left, reject value
    if (szLeft < ((int) nClusterCols * LcsMaxSzLeftError)) {
        // set szLeft to its previous value
        szLeft += sizeof(uint16_t);
        assert(szLeft >= 0);
        return false;
    }

    if (bFirstTimeInBatch) {
        // there is enough space to house the value, increment batch
        // value count
        batchDirs[column].nVal++;

        // check if nBits needs to change by comparing the value count
        // the change point count
        if (batchDirs[column].nVal == nextWidthChange[column]) {
            // calculate the next nBits value, and the count of values
            // for the next chane
            nBits[column] = calcWidth(batchDirs[column].nVal);
            nextWidthChange[column] = (1 << nBits[column]) + 1;
        }
    }

    return true;
}

// addValue() where the value must be added to the bottom of the page

bool LcsClusterNodeWriter::addValue(uint column, PBuffer pVal, uint16_t *oVal)
{
    uint16_t lastValOffset;
    int oldSzLeft = szLeft;
    uint szVal = attrAccessors[column].getStoredByteCount(pVal);

    // if we are in forced fixed compression mode,
    // see if the maximum record size in this batch has increased.
    // if so, adjust the szLeft based on the idea that each previous element
    // will now be taking more space
    if (bForceMode[column] == fixed) {
        if (szVal > maxValueSize[column]) {
            szLeft -= batchDirs[column].nVal
                * (szVal - maxValueSize[column]);
            maxValueSize[column] = szVal;
        }
    }

    // adjust szleft (upper bound on amount of space left in the block).
    // If we are in a forced fixed compression mode then we can calculate this
    // exactly.  If we are in "none" mode, then we calculate szLeft according to
    // variable mode compression (which should be an upper bound), and adjust it
    // later in pickCompressionMode
    if (bForceMode[column] == fixed) {
        szLeft -= maxValueSize[column];
    } else {
        // assume value is being added in variable mode
        // (note: this assumes that whenever we convert from
        // variable mode to compressed mode, the compressed mode will
        // take less space, for only in this case is szLeft an upper bound)
        szLeft -= (sizeof(uint16_t) + szVal) ;
    }

    // if there is not enough space left reject value
    if (szLeft < ((int) nClusterCols * LcsMaxSzLeftError)) {
        // set szLeft to its previous value
        szLeft = oldSzLeft;
        assert(szLeft >= 0);
        return false;
    }

    // otherwise, go ahead and add the value...

    lastValOffset = lastVal[column] - szVal;

    // there is enough space to house the value, increment batch value count
    batchDirs[column].nVal++;

    // check if nBits needs to change by comparing the value count
    // the change point count
    if (batchDirs[column].nVal == nextWidthChange[column]) {
        // calculate the next nBits value, and the count of values
        // for the next chane
        nBits[column] = calcWidth(batchDirs[column].nVal);
        nextWidthChange[column] = (1 << nBits[column]) + 1;
    }

    lastVal[column] = lastValOffset;

    // Save the value being inserted.  If we are building the
    // block in fixed mode then save the value into pValBank
    // rather than saving it into the block
    if (fixed == bForceMode[column]) {
        memcpy(pValBank[column] + lastValOffset, pVal, szVal);
    } else {
        memcpy(pBlock[column] + lastValOffset, pVal, szVal);
    }

    // return the block offset of the new value;
    *oVal = lastValOffset;

    nVal[column]++;

    return true;
}

void LcsClusterNodeWriter::undoValue(
    uint column, PBuffer pVal, bool bFirstInBatch)
{
    // pVal may be null if the value already exists, in which case, it wasn't
    // added to the value list.  However, if it was the first such value for
    // the batch, addValue was called to bump-up the batch value count
    // so we still need to call undoValue
    uint szVal =
        (pVal) ? attrAccessors[column].getStoredByteCount(pVal) : 0;

    // add back size subtracted for offset
    szLeft += (sizeof(uint16_t) + szVal) ;
    assert(szLeft >= 0);

    // If value was new to the batch, then adjust counters
    if (bFirstInBatch) {
        // decrement batch count
        batchDirs[column].nVal--;

        //reset nextWidthChange
        if (batchDirs[column].nVal == 0) {
            nextWidthChange[column] = 1;
        } else {
            // calculate the next nBits value, and the count of values
            // for the next chane
            nBits[column] = calcWidth(batchDirs[column].nVal);
            nextWidthChange[column] = (1 << nBits[column]) + 1;
        }
    }

    if (pVal) {
        // upgrage header
        lastVal[column] += szVal;
        nVal[column]--;
    }
}

void LcsClusterNodeWriter::putCompressedBatch(
    uint column, PBuffer pRows, PBuffer pBuf)
{
    uint        i, j, b;
    uint        iRow;
    uint        nByte;
    uint8_t     *pBit;
    uint16_t    *pOffs;
    PLcsBatchDir pBatch;

    WidthVec    w;      // bitVec width vector
    PtrVec      p;      // bitVec offsets
    uint        iV;     // number of bit vectors

    // pickCompressionMode() was called prior to putCompressedBatch,
    // and the following has been already done:
    // -- the batch descriptors were moved to the back of the batch
    // -- a batch descriptor for this batch has been placed in the batch
    //    directory
    // -- this->batch contains up to date info
    // -- the caller has copied nVal value offsets to the head of this batch

    // write to buffer values for rows over the 8 boundary if nrow is
    // greater then 8

    if (batchDirs[column].nRow > 8) {
        uint len;
        pOffs = (uint16_t *)(pBlock[column] + batchDirs[column].oVal);
        for (i = round8Boundary((uint32_t) batchDirs[column].nRow);
            i < batchDirs[column].nRow; i++, pBuf += batchDirs[column].recSize)
        {
            iRow = ((uint16_t *) pRows)[i];
            len =
                attrAccessors[column].getStoredByteCount(
                    pBlock[column] + pOffs[iRow]);
            memcpy(pBuf, pBlock[column] + pOffs[iRow], len);
        }
        batchDirs[column].nRow =
            round8Boundary((uint32_t) batchDirs[column].nRow);
    }

    // calculate the bit vector widthes, sum(w[i]) is nBits
    iV = bitVecWidth(nBits[column], w);

    // this is where the bit vectors start
    pBit = pBlock[column] + batchDirs[column].oVal
        + batchDirs[column].nVal*sizeof(uint16_t);

    // nByte are taken by the bit vectors, clear them befor OR-ing
    nByte = bitVecPtr(batchDirs[column].nRow, iV, w, p, pBit);
    memset(pBit, 0, nByte);

    for (j = 0, b = 0; j < iV ; j++) {
        switch (w[j]) {
        case 16:
            memcpy(p[j], pRows, batchDirs[column].nRow * sizeof(uint16_t));
            break;

        case 8:
            for (i = 0; i < batchDirs[column].nRow ; i++) {
                (p[j])[i] = (uint8_t)((uint16_t *) pRows)[i];
            }
            break;

        case 4:
            for (i = 0; i < batchDirs[column].nRow ; i++) {
                setBits(
                    p[j] + i / 2,
                    4,
                    (i % 2) * 4,
                    (uint16_t)(((uint16_t *) pRows)[i] >> b));
            }
            break;

        case 2:
            for (i = 0; i < batchDirs[column].nRow ; i++) {
                setBits(
                    p[j] + i / 4,
                    2,
                    (i % 4) * 2,
                    (uint16_t)(((uint16_t *) pRows)[i] >> b));
            }
            break;

        case 1:
            for (i = 0; i < batchDirs[column].nRow ; i++) {
                setBits(
                    p[j] + i / 8,
                    1,
                    (i % 8),
                    (uint16_t)(((uint16_t *)pRows)[i] >> b));
            }
            break;

        default:
                ;
        }
        b += w[j];
    }

    // put the batch in the batch directory
    pBatch = (PLcsBatchDir)(pBlock[column] + batchOffset[column]);
    pBatch[batchCount[column]] = batchDirs[column];
    batchCount[column]++;

    // reset the batch state
    batchDirs[column].mode = LCS_COMPRESSED;
    batchDirs[column].oLastValHighMark = lastVal[column];
    batchDirs[column].nValHighMark = nVal[column];
    batchDirs[column].nVal = 0;
    batchDirs[column].oVal = batchOffset[column];
    batchDirs[column].nRow = 0;

    // # of bits it takes to represent 0 values
    nBits[column] = 0;
    nextWidthChange[column] = 1 ;
}

void LcsClusterNodeWriter::putFixedVarBatch(
    uint column, uint16_t *pRows, PBuffer pBuf)
{
    uint        i;
    uint        batchRows;
    PBuffer     pVal;
    PLcsBatchDir pBatch;
    PBuffer     src;
    uint        batchRecSize;
    uint16_t    localLastVal;
    uint16_t    localoValBank;
    PBuffer     localpValBank, localpBlock;

    // The # of rows in a batch will be smaller then 8 or a multiple
    // of 8.  All but the last batch in a load will be greater then 8.
    // Round to nearest 8 boundary, unless this is the last batch (which
    // we determine by the nRows in the batch being less than 8).
    // If it turns it isn't the latch batch, then our caller (WriteBatch)
    // will roll back the whole batch and add it to a new block instead.
    batchRows = (batchDirs[column].nRow > 8)
        ? batchDirs[column].nRow & 0xfffffff8 : batchDirs[column].nRow;

    // the value destination
    pVal = pBlock[column] + batchDirs[column].oVal;
    if (batchDirs[column].mode == LCS_VARIABLE) {
        // For a variable mode, copy in the value offsets into the RI block.
        // The left over i.e. the rows over the highest 8 boundary will be
        // copied to pBuf
        memcpy(pVal, pRows, batchRows * sizeof(uint16_t));
    } else {
        // it's a fixed record batch
        assert(batchDirs[column].mode == LCS_FIXED);

        batchRecSize  = batchDirs[column].recSize;
        localLastVal  = lastVal[column];
        localpValBank = pValBank[column] + valBankStart[column];
        localoValBank = oValBank[column];
        localpBlock   = pBlock[column];

        // Copy the values themselves into the block.
        // The values are currently stored in pValBank
        for (i = 0; i < batchRows; i++) {
            // valueSource determines by the offset whether the value comes from
            // the bank of from the block
            src = valueSource(
                localLastVal, localpValBank, localoValBank,
                localpBlock, pRows[i]);
            uint len = attrAccessors[column].getStoredByteCount(src);
            memcpy(pVal, src, len);
            pVal += batchRecSize;
        }
    }

    // if forced fixed mode is true we need to periodically set it false, so
    // we can at least check if the data can be compressed
    if (bForceMode[column] != none) {
        if (forceModeCount[column] > 20) {
            bForceMode[column] = none;
            forceModeCount[column] = 0;
        }
    }

    batchRecSize  = batchDirs[column].recSize;
    localLastVal  = lastVal[column];
    localpValBank = pValBank[column] + valBankStart[column];
    localoValBank = oValBank[column];
    localpBlock   = pBlock[column];

    // copy the tail of the batch (the last nRow % 8 values) to pBuf
    pVal = pBuf;
    for (i = batchRows; i < batchDirs[column].nRow; i++) {
        // valueSource determines by the offset whether the value comes from
        // the bank or from the block. if the value bank is not used
        // valueSource will get all the values fron the block
        src = valueSource(
            localLastVal, localpValBank, localoValBank,
            localpBlock, pRows[i]);
        uint len = attrAccessors[column].getStoredByteCount(src);
        memcpy(pVal, src, len);
        pVal += batchRecSize;
    }

    if (pValBank[column]) {
        oValBank[column] = 0;
    }

    // Put batch descriptor in batch directory
    batchDirs[column].nRow = batchRows;
    pBatch = (PLcsBatchDir)(pBlock[column] + batchOffset[column]);
    pBatch[batchCount[column]] = batchDirs[column];

    // inc. batch count
    batchCount[column]++;

    // reset the batch state.  set batch back to compressed mode
    // unless the flag has been set that next
    switch (bForceMode[column]) {
    case none:
        batchDirs[column].mode = LCS_COMPRESSED;
        break;
    case fixed:
        batchDirs[column].mode = LCS_FIXED;
        break;
    case variable:
        batchDirs[column].mode = LCS_VARIABLE;
        break;
    default:
        assert(false);
    }
    batchDirs[column].oLastValHighMark = lastVal[column];
    batchDirs[column].nValHighMark = nVal[column];
    batchDirs[column].nVal = 0;
    batchDirs[column].oVal = batchOffset[column];
    batchDirs[column].nRow = 0;

    // # of bits it takes to represent 0 values
    nBits[column] = 0;
    nextWidthChange[column] = 1 ;

    maxValueSize[column] = 0;
}

void LcsClusterNodeWriter::pickCompressionMode(
    uint column, uint recSize, uint nRow, uint16_t **pValOffset,
    LcsBatchMode &compressionMode)
{
    uint        nByte;
    PLcsBatchDir pBatch;
    WidthVec    w;      // bitVec m_width vector
    uint        iV;     // number of bit vectors

    uint        szCompressed;   // size of the compressed batch
    uint        szVariable;     // size of the variable sized batch
    uint        szFixed;        // size of the fixed batch
    uint        szNonCompressed;
    uint        deltaVal;
    uint        batchRows;      // # of rows in the batch that is nRows rounded
                                // down to the nearest 8 boundary

    // update batch fields
    batchDirs[column].nRow = nRow;
    batchDirs[column].recSize = recSize;

    // calculate the size required for a compressed and sorted batch
    // by summing the spcae required for the value offsets, the bit vectors
    // and the values that were put in since the batch started

    // the # of rows in a batch will be smaller then 8 or a multiple
    // of 8. all but the last batch in a load will be greater then 8.
    batchRows = (nRow > 8) ? nRow & 0xfffffff8 : nRow;

    szCompressed = batchDirs[column].nVal * sizeof(uint16_t)
        + (nBits[column]*nRow + LcsMaxSzLeftError * 8) / 8
        + (batchDirs[column].oLastValHighMark - lastVal[column]);

    // the variable size batch does not have the bit vectors
    szVariable = batchDirs[column].nRow * sizeof(uint16_t)
        + (batchDirs[column].oLastValHighMark - lastVal[column]);

    // calculate the size required for the non compressed fixed mode
    // add max left overs to allow left overs to be added back
    uint    leftOverSize;
    leftOverSize = LcsMaxLeftOver * sizeof(uint16_t)
        + (3 * LcsMaxLeftOver + LcsMaxSzLeftError * 8) / 8
        + LcsMaxLeftOver * recSize;
    szFixed = nRow * recSize + leftOverSize;

    szNonCompressed = std::min(szFixed, szVariable);

    // Should we store this block in one of the non-compressed modes (fixed or
    // variable)?  We do this if either
    //    1) the non-compressed size is smaller than the compressed size
    // or 2) we built the block in forced uncompress mode
    //
    // test if the compressed size is bigger then the non compressed size

    if ((fixed == bForceMode[column] || variable == bForceMode[column])
        || szCompressed > szNonCompressed)
    {
        // switch to one of the noncompressed modes
        *pValOffset = NULL;
        batchDirs[column].nVal = 0;

        forceModeCount[column]++;

        // If we are storing it in fixed mode...
        if (fixed == bForceMode[column] || szNonCompressed == szFixed) {
            // batch will be stored in fixed mode
            // change mode
            batchDirs[column].mode = LCS_FIXED;

            // Are we switching from variable mode to fixed mode?
            if (bForceMode[column] != fixed) {
                // We are going to store the batch in fixed mode.  But currently
                // it is saved in variable mode.  Save the value pointers into
                // pValBank and we will use them later to help in conversion

                // number of bytes taken by value over the batch high mark point
                deltaVal = batchDirs[column].oLastValHighMark - lastVal[column];

                // save the values obtained while this batch was in
                // variable mode
                if (deltaVal) {
                    memcpy(
                        pValBank[column],
                        pBlock[column] + lastVal[column],
                        deltaVal);
                }

                valBankStart[column] = 0;

                // mark that for the next few times, automatically go to
                // fixed mode without checking if it is the best
                bForceMode[column] = fixed;

                // Adjust szLeft since we have freed up some space in
                // the block
                assert(szVariable >= szFixed);
                szLeft += (szVariable - szFixed);
                assert(szLeft >= 0);
            } else {
                valBankStart[column] = lastVal[column];
            }

            // Reclaim the space at the bottom of the block that
            // used to hold the values

            // first offset included in the bank
            oValBank[column] = lastVal[column];
            lastVal[column] = batchDirs[column].oLastValHighMark;
            nVal[column] = batchDirs[column].nValHighMark;

            // the number of bytes taken by the batch
            nByte = batchRows * batchDirs[column].recSize;

        } else {
            // batch will be stored in variable mode

            batchDirs[column].mode = LCS_VARIABLE;

            // the number of bytes taken by the batch
            nByte = batchRows*sizeof(uint16_t);

            // mark that for the next few times, automatically go to
            // fixed mode without checking if it is the best
            bForceMode[column] = variable;
        }
    } else {
        // batch will be stored in compressed mode
        // values will be put at the start of the new batch

        *pValOffset = (uint16_t *)(pBlock[column] + batchOffset[column]);

        // calculate the bit vector widthes
        iV = bitVecWidth(nBits[column], w);

        // nByte is the # bytes taken by the batch, it is the sum of
        // the bit vectors size and the value offsets
        nByte = sizeofBitVec(batchRows, iV, w)
            + batchDirs[column].nVal * sizeof(uint16_t);

        // Adjust szLeft since we have freed up some space in the block
        assert(szVariable >= szCompressed);
        szLeft += (szVariable - szCompressed);
        assert(szLeft >= 0);
    }

    compressionMode = batchDirs[column].mode;

    // Slide down the batch directories to make room for new batch data (batch
    // directories occur after the batches).
    pBatch = (PLcsBatchDir)(pBlock[column] + batchOffset[column] + nByte);
    memmove(
        pBatch,
        pBlock[column] + batchOffset[column],
        batchCount[column] * sizeof(LcsBatchDir));

    // adjust szLeft for the space used by the next batch to reflect only
    // its batch directory
    szLeft -= sizeof(LcsBatchDir);
    if (szLeft < 0) {
        szLeft = 0;
    }

    // batchDirs[column].oVal points to where the batch dir used to be,
    // and this is where the batch records will start
    batchDirs[column].oVal = batchOffset[column];

    // set batchOffset[column] to point to the start of the batch
    // directores (if we have another batch then this will become the
    // offset of the new batch)
    batchOffset[column] = (batchOffset[column] + nByte);
}

// myCopy: like memcpy(), but optimized for case where source
// and destination are the same (ie, when we have a single column
// cluster and we are copying from the index block to the temporary
// blocks, this will do nothing because the temp block just points
// back to the index block)
void myCopy(void* pDest, void* pSrc, uint sz)
{
    if (pDest == pSrc) {
        return;
    } else {
        memcpy(pDest, pSrc, sz);
    }
}

RecordNum LcsClusterNodeWriter::moveFromIndexToTemp()
{
    PLcsBatchDir pBatch;
    boost::scoped_array<uint16_t> batchDirOffset;
    uint16_t loc;
    uint column;
    uint batchCount = pHdr->nBatch / nClusterCols;
    uint b;

    batchDirOffset.reset(new uint16_t[pHdr->nBatch]);

    // First move the values
    //
    // copy values from index for all columns starting with the
    // 1st column in cluster.
    for (column = 0; column < nClusterCols; column++) {
        uint sz = firstVal[column] - lastVal[column];
        loc = (uint16_t) (szBlock - sz);
        myCopy(pBlock[column] + loc, pIndexBlock + lastVal[column], sz);

        // adjust lastVal and firstVal to offset in temporary block
        lastVal[column]  = loc;
        firstVal[column] = (uint16_t) szBlock;
    }

    // Next move the batches

    pBatch = (PLcsBatchDir)(pIndexBlock + pHdr->oBatch);
    for (column = 0; column < nClusterCols; column++) {
        uint i;
        loc = hdrSize;

        // move every batch for this column
        for (b = column, i = 0; i < batchCount; i++, b = b + nClusterCols) {
            uint16_t    batchStart = loc;

            if (pBatch[b].mode == LCS_COMPRESSED) {
                uint8_t     *pBit;
                WidthVec    w;      // bitVec m_width vector
                PtrVec      p;      // bitVec offsets
                uint        iV;     // # of bit vectors
                uint        sizeOffsets, nBytes;

                //copy offsets
                sizeOffsets =  pBatch[b].nVal * sizeof(uint16_t);
                myCopy(
                    pBlock[column] + loc, pIndexBlock + pBatch[b].oVal,
                    sizeOffsets);

                // step past offsets
                loc = (uint16_t) (loc + sizeOffsets);

                // calculate the bit vector widthes
                iV = bitVecWidth(calcWidth(pBatch[b].nVal), w);

                // this is where the bit vectors start
                pBit = pIndexBlock + pBatch[b].oVal + sizeOffsets;

                // nByte are taken by the bit vectors
                nBytes = bitVecPtr(pBatch[b].nRow, iV, w, p, pBit);

                myCopy(pBlock[column] + loc, pBit, nBytes);

                // step past bit vectors
                loc = (uint16_t) (loc + nBytes);
            } else if (pBatch[b].mode == LCS_VARIABLE) {
                uint        sizeOffsets;

                sizeOffsets = pBatch[b].nRow * sizeof(uint16_t);

                // variable size record batch
                myCopy(
                    pBlock[column] + loc, pIndexBlock + pBatch[b].oVal,
                    sizeOffsets);

                // step past offsets
                loc = (uint16_t) (loc + sizeOffsets);
            } else {
                // fixed mode batch
                uint sizeFixed;

                sizeFixed =  pBatch[b].nRow * pBatch[b].recSize;
                // fixed size record batch
                myCopy(
                    pBlock[column] + loc, pIndexBlock + pBatch[b].oVal,
                    sizeFixed);

                //step past fixed records
                loc = (uint16_t) (loc + sizeFixed);
            }

            // set offset where values start in temp block
            batchDirOffset[b] = batchStart;
        }

        // move batch directories for this column

        uint16_t  dirLoc;
        b = column;
        dirLoc = loc;
        batchOffset[column] = dirLoc;

        // move every batch for this column
        for (i = 0; i < batchCount; i++) {
            PLcsBatchDir pTempBatch = (PLcsBatchDir)(pBlock[column] + dirLoc);
            myCopy(pTempBatch, &pBatch[b], sizeof(LcsBatchDir));

            pTempBatch->oVal = batchDirOffset[b];
            // increment to next batch and next location in temp block
            b = b + nClusterCols;
            dirLoc += sizeof(LcsBatchDir);
        }
    }

    // compute the number of rows on the page
    pBatch = (PLcsBatchDir)(pIndexBlock + pHdr->oBatch);
    RecordNum nrows = 0;
    for (b = 0; b < pHdr->nBatch; b = b + nClusterCols) {
        nrows += pBatch[b].nRow;
    }

    batchDirOffset.reset();
    return nrows;
}

void LcsClusterNodeWriter::moveFromTempToIndex()
{
    PLcsBatchDir pBatch;
    uint        sz, numBatches = batchCount[0];
    uint16_t    offset, loc;
    uint        column, b;

    // Copy values from temporary blocks for all columns starting with the
    // 1st column in cluster.

    for (offset = (uint16_t) szBlock, column = 0; column < nClusterCols;
        column++)
    {
        sz = szBlock - lastVal[column];
        myCopy(
            pIndexBlock + (offset - sz), pBlock[column] + lastVal[column], sz);

        //  set delta value to subtract from offsets to get relative offset
        delta[column] = (uint16_t)(szBlock - offset);

        // adjust firstVal and lastVal in the leaf block header to appropriate
        // offsets in index block (currently base on offsets in temporary block)
        firstVal[column] = offset;
        offset = (uint16_t) (offset - sz);
        lastVal[column] = offset;
    }

    // copy batch descriptors (which point to the batches)

    for (loc =  hdrSize, b = 0; b < numBatches; b++) {
        for (column = 0; column < nClusterCols; column++) {
            uint16_t    batchStart = loc;

            pBatch = (PLcsBatchDir)(pBlock[column] + batchOffset[column]);

            if (pBatch[b].mode == LCS_COMPRESSED) {
                uint8_t     *pBit;
                WidthVec    w;      // bitVec m_width vector
                PtrVec      p;      // bitVec offsets
                uint        iV;     // # of bit vectors
                uint        sizeOffsets, nBytes;

                sizeOffsets =  pBatch[b].nVal * sizeof(uint16_t);

                // first copy offsets then bit vectors
                myCopy(
                    pIndexBlock + loc, pBlock[column] + pBatch[b].oVal,
                    sizeOffsets);

                // step past offsets
                loc = (uint16_t) (loc + sizeOffsets);

                // calculate the bit vector widthes
                iV = bitVecWidth(calcWidth(pBatch[b].nVal), w);

                // this is where the bit vectors start in temporary block
                pBit = pBlock[column] + pBatch[b].oVal + sizeOffsets;

                // nByte are taken by the bit vectors
                nBytes = bitVecPtr(pBatch[b].nRow, iV, w, p, pBit);

                myCopy(pIndexBlock + loc, pBit, nBytes);

                // step past bit vectors
                loc = (uint16_t)(loc + nBytes);

            } else if (pBatch[b].mode == LCS_VARIABLE) {
                uint        sizeOffsets;

                sizeOffsets =  pBatch[b].nRow * sizeof(uint16_t);

                // variable size record batch
                myCopy(
                    pIndexBlock + loc, pBlock[column] + pBatch[b].oVal,
                    sizeOffsets);

                // step past offsets
                loc = (uint16_t) (loc + sizeOffsets);
            } else {
                // Fixed mode
                uint sizeFixed;

                sizeFixed =  pBatch[b].nRow * pBatch[b].recSize;
                // fixed size record batch
                myCopy(
                    pIndexBlock + loc, pBlock[column] + pBatch[b].oVal,
                    sizeFixed);

                //step past fixed records
                loc = (uint16_t) (loc + sizeFixed);
            }

            // set offset where values start in indexBlock
            pBatch[b].oVal = batchStart;
        }
    }

    //adjust batch count in leaf block header
    pHdr->nBatch = nClusterCols * numBatches;

    // start batch directory at end of last batch
    pHdr->oBatch = loc;

    // copy batch directories
    for (b = 0; b < numBatches; b++) {
        for (column = 0; column < nClusterCols; column++) {
            pBatch = (PLcsBatchDir)(pBlock[column] + batchOffset[column]);
            myCopy(pIndexBlock + loc, &pBatch[b], sizeof(LcsBatchDir));
            loc += sizeof(LcsBatchDir);
        }
    }

    if (isTracingLevel(TRACE_FINE)) {
        FENNEL_TRACE(
            TRACE_FINE, "Calling ClusterDump from moveFromTempToIndex");
        clusterDump->dump(opaqueToInt(clusterPageId), pHdr, szBlock);
    }
}

void LcsClusterNodeWriter::allocArrays()
{
    // allocate arrays only if they have not been allocated already
    if (!arraysAllocated) {
        arraysAllocated = true;

        batchDirs.reset(new LcsBatchDir[nClusterCols]);

        pValBank.reset(new PBuffer[nClusterCols]);

        // allocate larger buffers for the individual pages in the value bank

        attrAccessors.reset(new UnalignedAttributeAccessor[nClusterCols]);

        for (uint col = 0; col < nClusterCols; col++) {
            bufferLock.allocatePage();
            pValBank[col] = bufferLock.getPage().getWritableData();
            // Similar to what's done in external sorter, we rely on the fact
            // that the underlying ScratchSegment keeps the page pinned for us.
            // The pages will be released when all other pages associated with
            // the ScratchSegment are released.
            bufferLock.unlock();

            attrAccessors[col].compute(colTupleDesc[col]);
        }

        valBankStart.reset(new uint16_t[nClusterCols]);

        forceModeCount.reset(new uint[nClusterCols]);

        bForceMode.reset(new ForceMode[nClusterCols]);

        oValBank.reset(new uint16_t[nClusterCols]);

        batchOffset.reset(new uint16_t[nClusterCols]);

        batchCount.reset(new uint[nClusterCols]);

        nBits.reset(new uint[nClusterCols]);

        nextWidthChange.reset(new uint[nClusterCols]);

        maxValueSize.reset(new uint[nClusterCols]);
    }

    memset(valBankStart.get(), 0, nClusterCols * sizeof(uint16_t));
    memset(forceModeCount.get(), 0, nClusterCols * sizeof(uint));
    memset(bForceMode.get(), 0, nClusterCols * sizeof(ForceMode));
    memset(oValBank.get(), 0, nClusterCols * sizeof(uint16_t));
    memset(batchOffset.get(), 0, nClusterCols * sizeof(uint16_t));
    memset(batchCount.get(), 0, nClusterCols * sizeof(uint));
    memset(nBits.get(), 0, nClusterCols * sizeof(uint));
    memset(nextWidthChange.get(), 0, nClusterCols * sizeof(uint));
    memset(maxValueSize.get(), 0, nClusterCols * sizeof(uint));
}

FENNEL_END_CPPFILE("$Id$");

// End LcsClusterNodeWriter.cpp


text:
/*
 * File:   AICpanda.cpp
 * Author: Corrado Pezzato, CoR, TU Deft
 *
 * Created on September 4th, 2019
 *
 * This node allows to control the 7DOF Franka Emika Panda robot arm through
 * Active Inference
 *
 * The control is in joint space, sending torque commands and using position and
 * velocity feedback. The robot moves to the desired goal from the initial pose.
 *
 * ---> Usage:: ./<folder_code>/AICpanda.cpp <fci-ip>
 *
 */

#include <iostream>
#include <array>
#include <franka/robot.h>
#include <franka/duration.h>
#include <franka/exception.h>
#include <Eigen/Core>
#include <Eigen/Dense>

#include "examples_common.h"

int main(int argc, char** argv) {
 // Check if the user provided the hostname or IP
 if (argc != 2) {
   std::cerr << "Usage: " << argv[0] << " <robot-hostname>" << std::endl;
   return -1;
 }
 // If hostname or IP provided, try to connect to the robot
 try {

   // Define an object for the Panda to handle the control signals and read
   // the sensors. Using the method robot.control() you read the
   // joint values at 1kHz
   franka::Robot robot(argv[1]);
   setDefaultBehavior(robot);

   // Set collision behavior
   robot.setCollisionBehavior({{100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0}},
                              {{100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0}},
                              {{100.0, 100.0, 100.0, 100.0, 100.0, 100.0}},
                              {{100.0, 100.0, 100.0, 100.0, 100.0, 100.0}});

   // Define controller
   /////////// Controller parameters /////////////
   // Auxiliary variables
   double time = 0.0;
   double maxTime = 100.0;
   // Variances for internal belief and sensory input, learning rates, integration step
   double var_q, var_qdot, var_mu, var_muprime, k_mu, k_a, h;
   // Precision matrices, diagonal matrices with the inverce of the variance
   Eigen::Matrix<double, 7, 7> SigmaP_yq0, SigmaP_yq1, SigmaP_mu, SigmaP_muprime;
   // Belief about the states and their derivatives mu, mu', mu'', joint states, desired value, control
   Eigen::Matrix<double, 7, 1> mu, mu_p, mu_pp, mu_dot, mu_dot_p, mu_dot_pp, jointPos, jointVel, mu_d, u;

   // Set desired goal
   // Center
   mu_d << 0.0345169,-0.167844,-0.000332711,-1.74526,0.00905814,1.56589,0.824567;

   // Right
   //mu_d << 0.15086,0.235858,-0.458875,-2.44786,0.0599695,2.6241,0.450365;

   //////////// Controller tuning //////////////
   // Variances associated with the beliefs and the sensory inputs
   var_mu = 5.0;
   var_muprime = 10.0;
   var_q = 1;
   var_qdot = 1;

   // Learning rates for the gradient descent
   k_mu = 11.67;
   k_a = 200;

   // Integration step
   h = 0.001;

   // Precision matrices (first set them to zero then populate the diagonal)
   SigmaP_yq0 = Eigen::Matrix<double, 7, 7>::Zero();
   SigmaP_yq1 = Eigen::Matrix<double, 7, 7>::Zero();
   SigmaP_mu = Eigen::Matrix<double, 7, 7>::Zero();
   SigmaP_muprime = Eigen::Matrix<double, 7, 7>::Zero();
   //////////////////////////////////////////////

   // Read robot initial state to set the initial belief
   franka::RobotState initial_state = robot.readOnce();
   // Initialization
   for( int i = 0; i < 7; i = i + 1 ) {
     SigmaP_yq0(i,i) = 1/var_q;
     SigmaP_yq1(i,i) = 1/var_qdot;
     SigmaP_mu(i,i) = 1/var_mu;
     SigmaP_muprime(i,i) = 1/var_muprime;
     // Internal belief starting from initial pose
     jointPos(i) = initial_state.q[i];
     jointVel(i) = initial_state.dq[i];
   }
   // Initial belief
   mu = jointPos;
   mu_p = jointVel;
   mu_pp << 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0;
   // Initial control actions
   u << 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0;

   // Controller callback for RT loop
   auto my_AIC_controller_callback = [&](const franka::RobotState& robot_state,
                                     franka::Duration period) -> franka::Torques {
     time += period.toSec();
     // Save the robot state for algebric manipulation
     for( int i = 0; i < 7; i = i + 1 ) {
       // Set current sensory input
       jointPos(i) = robot_state.q[i];
       jointVel(i) = robot_state.dq[i];
     }

     /////////// Active inference ///////////
     // Belief update
     mu_dot = mu_p - k_mu*(-SigmaP_yq0*(jointPos-mu)+SigmaP_mu*(mu_p+mu-mu_d));
     mu_dot_p = mu_pp - k_mu*(-SigmaP_yq1*(jointVel-mu_p)+SigmaP_mu*(mu_p+mu-mu_d)+SigmaP_muprime*(mu_pp+mu_p));
     mu_dot_pp = - k_mu*(SigmaP_muprime*(mu_pp+mu_p));

     mu = mu + h*mu_dot;             // Belief about the position
     mu_p = mu_p + h*mu_dot_p;       // Belief about motion of mu
     mu_pp = mu_pp + h*mu_dot_pp;    // Belief about motion of mu'

     // Compute control actions
     u = u-h*k_a*(SigmaP_yq1*(jointVel-mu_p)+SigmaP_yq0*(jointPos-mu));

    // Build control signal and send it to the robot
    franka::Torques tau_cmd = {{u(0), u(1), u(2), u(3), u(4), u(5), u(6)}};
    // Set torques to zero for testing
    // franka::Torques tau_cmd = {{0, 0, 0, 0, 0, 0, 0}};

    if (time >= maxTime) {
      std::cout << std::endl << "Finished motion, shutting down demo" << std::endl;
      return franka::MotionFinished(tau_cmd);
    }
    return tau_cmd;
   };
   std::cout << "WARNING: Make sure sure that the robot has enough space to move. Keep in mind that "
                "collision thresholds are set to high values."
             << std::endl
             << "Press Enter to start the demo..." << std::endl;
   std::cin.ignore();
   // start real-time control loop
   robot.control(my_AIC_controller_callback);
   // Catch the exemptions if any connection errors occurred
 } catch (franka::Exception const& e) {
  std::cout << e.what() << std::endl;
     return -1;
   }
  return 0;
}