PeerStreamer

/*

  Copyright 2003 John Knottenbelt

  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, USA.

*/

#include "scheduler.h"

#include "select.h"

#include "connection.h"

#include "connectionmanager.h"

#include "tunermanager.h"

#include "faketunermanager.h"

#include "utils.h"

#include "stringutil.h"

#include <sys/types.h>

#include <sys/stat.h>

#include <fcntl.h>

#include <cerrno>

#include <cstring>

#include <fstream>

#include <ctime>

#include <algorithm>

Scheduler::Scheduler(const std::string &dbfile, TunerManager *tm, ConnectionManager *cm)

  : lastJob(0), tm(tm), cm(cm), dbfile(dbfile)

{

  readDB();

}

Scheduler::~Scheduler()

{

}

void Scheduler::add( Recording &r )

{

  r.setJob(lastJob++);

  // Insert into the list sorted on start time

  for (List::iterator i = recordings.begin(); i != recordings.end(); i++) {

    if (r.getStartTime() < i->getStartTime()) {

      recordings.insert(i, r);

      writeDB();

      return;

    }

  }

  recordings.push_back(r);

  writeDB();

}

void Scheduler::readDB()

{

  // NB: We assume that the database is stored in sorted order.

  //     (this is true if it was written by writeDB)

  std::ifstream i(dbfile.c_str());

  if (!i) {

    std::cerr << "Warning: failed to open database file " << dbfile << " for reading:\n"

              << strerror(errno) << "\n";

    writeDB();

  }

  Recording r;

  while (i >> r) {

    recordings.push_back(r);

    lastJob = std::max(r.getJob() + 1, lastJob);

  }  

}

void Scheduler::writeDB() const

{

  // Write the db file

  std::ofstream o(dbfile.c_str());

  if (!o) {

    std::cerr << "Warning: failed to open database file " << dbfile << " for writing:\n"

              << strerror(errno) << "\n";

    return;

  }

  for (List::const_iterator i = recordings.begin(); i != recordings.end(); i++) 

    o << *i << "\n";

  o.close();

}

void Scheduler::addSelectFDs(Select &s) const

{

  long tv_sec, tv_usec;

  if (timeUntilNextRecording(tv_sec, tv_usec))

    s.addAlarm(tv_sec, tv_usec);

}

bool Scheduler::isReady(const Select &) const

{

  long tv_sec, tv_usec;

  if (timeUntilNextRecording(tv_sec, tv_usec))

    return tv_sec == 0 && tv_usec == 0;

  return false;

}

bool Scheduler::timeUntilNextRecording( long &tv_sec, long &tv_usec ) const

{

  time_t now = time(NULL);

  if (recordings.empty())

    return false;

  const Recording &r = *recordings.begin();

  tv_usec = 0;

  tv_sec = std::max(0L, r.getStartTime() - now);

  return true;

}

void Scheduler::process()

{

  long wait_sec, wait_usec;

  if (timeUntilNextRecording(wait_sec, wait_usec) &&

      wait_sec == 0 && wait_usec == 0) {

    Recording r = *recordings.begin();

    std::cout << "Scheduled recording\n"

              << r << std::endl;

    startRecording(r);

    recordings.pop_front();

    writeDB();

  }

}

void Scheduler::startRecording(const Recording &r)

{

  if (!accessOk(r.getPath(), r.getUid())) {

    std::cerr << "Insufficient access to " << r.getPath() << "\n";

    return;

  }

  // Don't overwrite existing files

  std::string path = r.getPath();

  int n = 0;

  while (access(path.c_str(), F_OK) == 0)

    path = r.getPath() + "." + toString(++n);

  int openMode = O_WRONLY|O_CREAT|O_TRUNC|O_LARGEFILE;

  // Don't bother caching if possible

#ifdef O_STREAMING

  openMode |= O_STREAMING;

#endif

  int dataFD = open(r.getPath().c_str(), openMode, 0644);

  if (dataFD == -1) {

    std::cerr << "Failed to open file " << r.getPath() << " for writing: " 

              << strerror(errno) << "\n";

  }

  Connection *c = new Connection( tm, dataFD );

  c->setStopTime( r.getStopTime() );

  c->tune(r.getType(), r.getChannel(), r.getConvertToPS(), r.getPriority());

  cm->add(c);

}

bool Scheduler::overlaps(const List &l, const Recording &r) const 

{

  for (List::const_iterator i = l.begin(); i != l.end(); i++)

    if (i->getStartTime() <= r.getStartTime() && 

        r.getStartTime() <= i->getStopTime())

      return true;

  return false;

}

void Scheduler::checkOverlappingBlock(const List &l, StringList &interruptions) const

{

  FakeTunerManager fakeTM(*tm);

  int devNull = open("/dev/null", O_WRONLY);

  typedef std::list<Connection *> ConnectionList;

  typedef std::map<Connection *, Recording> ConnectionMap;

  ConnectionList connections;

  ConnectionMap programs;

  // Here would be the place to set up any existing connections

  // We assume that the devices are unoccupied. This might

  // not be true, if for example the scheduled program is

  // due soon and a channel is currently connected

  // at high priority.

  for (List::const_iterator i = l.begin(); i != l.end(); i++) {

    const Recording &r = *i;

    int now = r.getStartTime();

    // Expire old connections

    for (ConnectionList::iterator j = connections.begin(); j != connections.end(); ) {

      Connection *e = *j;

      if (now > e->getStopTime()) {

        delete e;

        j = connections.erase(j);

      }

      else j++;

    }

    // Add in a new connection

    Connection *c = new Connection( &fakeTM, dup(devNull) );

    programs[c] = r;

    c->setStopTime( r.getStopTime() );

    if (!c->tune(r.getType(), r.getChannel(), r.getConvertToPS(), r.getPriority())) {

      interruptions.push_back( "Failed to start recording: " + r.toString() );

      delete c;

    }

    else {

      // Check to see if the new connection displaced any of the

      // others

      for (ConnectionList::iterator j = connections.begin(); j != connections.end(); ) {

        Connection *e = *j;

        if (e->getInterrupted()) {

          interruptions.push_back( "Recording " + r.toString() + " interrupts " + programs[e].toString() );

          delete e;

          j = connections.erase(j);

        }

        else j++;

      }

      // Add the new connection to the list

      connections.push_back(c);

    }

  }

  // Terminate the last connections

  for (ConnectionList::iterator j = connections.begin(); j != connections.end(); j++)

    delete *j;

  ::close(devNull);

}

void Scheduler::checkOverlaps(StringList &interruptions) const

{

  /** Find overlapping schedules and check that the it is possible

      to schedule them */

  for (List::const_iterator i = recordings.begin(); i != recordings.end(); ) {

    const Recording &cur = *i;

    i++;

    if (i == recordings.end())

      break;

    List overlapping;

    overlapping.push_back(cur);

    while (i != recordings.end() && overlaps(overlapping, *i)) {

      overlapping.push_back(*i);

      i++;

    }

    if (overlapping.size() > 1) {

      checkOverlappingBlock(overlapping, interruptions);

    }

  }

}

void Scheduler::getSchedule(StringList &s) const

{

  for (List::const_iterator i = recordings.begin(); i != recordings.end(); i++)

    s.push_back((*i).toString());

}

bool Scheduler::remove(int job)

{

  for (List::iterator i = recordings.begin(); i != recordings.end(); ) {

    if ((*i).getJob() == job) {

      recordings.erase(i);

      return true;

    }

    else i++;

  }

  return false;

}

bool Scheduler::getNextJobTime( time_t &time ) const

{

  if (recordings.empty())

    return false;

  const Recording &r = *recordings.begin();

  time = r.getStartTime();

  return true;

}