active-objects.h

Go to the documentation of this file.

/****
 Active Objects v4.2
 Copyright (C) 2006-2010 Werner Van Belle
 http://active.yellowcouch.org/

 This program is free software; you can redistribute it and/or modify
 it under the terms of the GNU General Public License as published by
 the Free Software Foundation; either version 2 of the License, or
 (at your option) any later version.

 This program is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 GNU General Public License for more details.

 You should have received a copy of the GNU General Public License
 along with this program; if not, write to the Free Software
 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
****/
#ifndef __loaded__active_objects_h__
#define __loaded__active_objects_h__
using namespace std;
#line 1 "active-objects.h++"
#include <cstdio>
#include <iostream>
#include <string>
#include <pthread.h>
#include <stdlib.h>
#include <assert.h>
#include <vector>
#include <queue>
#include <deque>
#include <time.h>
#include "ao-scheduler.h"
#include "ao-pool.h"
using namespace std;

typedef enum{RevisitLater, Revisit, RevisitAfterIncoming, Done, Interrupt} 
  elementResult;

template <class message>
class ActiveObject: public Lock, protected Runnable
{
 private:
  deque<message> incoming;
 protected:
  deque<message> handling;
 protected:
  ActiveObject(string name): Runnable(name)
  {
    active = false;
    changed = false;
    assert(aoPool);
    scheduler = aoPool->sunrise(name);
    assert(scheduler);
  };
public:
  void push(message d)
  {
    Synchronized(this);
    push_internal(d);
  }
  void push(vector<message> v)
  {
    Synchronized(this);
    for(int i = 0 ; i<v.size();i++)
      push_internal(v[i]);
  }
  virtual ~ActiveObject()
  {
    if (active)
      cerr << "Active Object " << name << " is still active\n";
    if (scheduler)
      cerr << "Active Object " << name << " can still be activated\n";
  }
private:
  void push_internal(message d)
  {
    assert(locked);
    incoming.push_back(d);
    activate();
  }
 protected:
  Scheduler * scheduler;
 public:
  virtual bool terminated()
  {
    return scheduler==NULL;
  }
 private:
  Lock updating_state;
  volatile bool active;
  volatile bool changed;
  void activate()
    {
      if (!scheduler) return;
      {
        Synchronized(updating_state);
        changed = true;
        if (active) return;
        active = true;
      }
      assert(scheduler);
      scheduler->start(this);
    }
protected:
  elementResult deactivate()
  {
#ifdef DEBUG_ACTIVATION
    cerr << "Deactivating " << name << endl;
    cerr.flush();
#endif 
    assert(scheduler);
    scheduler = false;
    if (aoPool)
      aoPool->sunset(name);
    return Done;
  }
private:
  bool step()
    {
      bool transfer_needed = false;
      {
        Synchronized(updating_state);
        if (changed)
          {
            transfer_needed = true;
            changed = false;
          }
      }
      if (transfer_needed)
        {
          Synchronized(this);
          while(!incoming.empty())
            {
              handling.push_back(incoming.front());
              incoming.pop_front();
            }
        }
      if (handle() && scheduler)
          return true;
      {
        Synchronized(updating_state);
        if (changed) return true;
        active = false;
        return false;
      }
    }
protected:
  virtual bool run(bool handle_as_many_as_possible)
    {
      if (handle_as_many_as_possible)
        {
          while(step()) {};
          return false;
        }
      else
        return step();
    }
  virtual bool handle()
    {
      if (handling.empty()) 
        return false;
      message f = handling.front();
      elementResult hr = handle(f);
      switch(hr)
        {
        case RevisitLater:
          handling.push_back(f);
          handling.pop_front();
          // no deletion of the message
          break;
        case Done:
          handling.pop_front();
          // deletion of message when done
          delete f;
          break;
        case Revisit:
          break;
        case Interrupt:
          return false;
        case RevisitAfterIncoming:
          push(f);
          handling.pop_front();
          break;
        default:
          assert(0);
        }
      bool t=!handling.empty();
      return t;
    }

  virtual elementResult handle(message)
  {
    cerr << name << " lacks a handle(message) routine\n";
    assert(0);
    return Interrupt;
  }
};
#endif // __loaded__active_objects_h__

---