Spaces:

suricodes
/

hindi-sindhi-docker

Paused

hindi-sindhi-docker / mosesdecoder /util /stream /chain.cc

Upload folder using huggingface_hub

fd49381 verified about 2 years ago

4.45 kB

#include "util/stream/chain.hh"

#include "util/stream/io.hh"

#include "util/exception.hh"
#include "util/pcqueue.hh"

#include <cstdlib>
#include <new>
#include <iostream>
#include <stdint.h>

namespace util {
namespace stream {

ChainConfigException::ChainConfigException() throw() { *this << "Chain configured with "; }
ChainConfigException::~ChainConfigException() throw() {}

Thread::~Thread() {
  thread_.join();
}

void Thread::UnhandledException(const std::exception &e) {
  std::cerr << e.what() << std::endl;
  abort();
}

void Recycler::Run(const ChainPosition &position) {
  for (Link l(position); l; ++l) {
    l->SetValidSize(position.GetChain().BlockSize());
  }
}

const Recycler kRecycle = Recycler();

Chain::Chain(const ChainConfig &config) : config_(config), complete_called_(false) {
  UTIL_THROW_IF(!config.entry_size, ChainConfigException, "zero-size entries.");
  UTIL_THROW_IF(!config.block_count, ChainConfigException, "block count zero");
  UTIL_THROW_IF(config.total_memory < config.entry_size * config.block_count, ChainConfigException, config.total_memory << " total memory, too small for " << config.block_count << " blocks of containing entries of size " << config.entry_size);
  // Round down block size to a multiple of entry size.
  block_size_ = config.total_memory / (config.block_count * config.entry_size) * config.entry_size;
}

Chain::~Chain() {
  Wait();
}

ChainPosition Chain::Add() {
  if (!Running()) Start();
  PCQueue<Block> &in = queues_.back();
  queues_.push_back(new PCQueue<Block>(config_.block_count));
  return ChainPosition(in, queues_.back(), this, progress_);
}

Chain &Chain::operator>>(const WriteAndRecycle &writer) {
  threads_.push_back(new Thread(Complete(), writer));
  return *this;
}

Chain &Chain::operator>>(const PWriteAndRecycle &writer) {
  threads_.push_back(new Thread(Complete(), writer));
  return *this;
}

void Chain::Wait(bool release_memory) {
  if (queues_.empty()) {
    assert(threads_.empty());
    return; // Nothing to wait for.
  }
  if (!complete_called_) CompleteLoop();
  threads_.clear();
  for (std::size_t i = 0; queues_.front().Consume(); ++i) {
    if (i == config_.block_count) {
      std::cerr << "Chain ending without poison." << std::endl;
      abort();
    }
  }
  queues_.clear();
  progress_.Finished();
  complete_called_ = false;
  if (release_memory) memory_.reset();
}

void Chain::Start() {
  Wait(false);
  if (!memory_.get()) {
    // Allocate memory.
    assert(threads_.empty());
    assert(queues_.empty());
    std::size_t malloc_size = block_size_ * config_.block_count;
    memory_.reset(MallocOrThrow(malloc_size));
  }
  // This queue can accomodate all blocks.
  queues_.push_back(new PCQueue<Block>(config_.block_count));
  // Populate the lead queue with blocks.
  uint8_t *base = static_cast<uint8_t*>(memory_.get());
  for (std::size_t i = 0; i < config_.block_count; ++i) {
    queues_.front().Produce(Block(base, block_size_));
    base += block_size_;
  }
}

ChainPosition Chain::Complete() {
  assert(Running());
  UTIL_THROW_IF(complete_called_, util::Exception, "CompleteLoop() called twice");
  complete_called_ = true;
  return ChainPosition(queues_.back(), queues_.front(), this, progress_);
}

Link::Link() : in_(NULL), out_(NULL), poisoned_(true) {}

void Link::Init(const ChainPosition &position) {
  UTIL_THROW_IF(in_, util::Exception, "Link::Init twice");
  in_ = position.in_;
  out_ = position.out_;
  poisoned_ = false;
  progress_ = position.progress_;
  in_->Consume(current_);
}

Link::Link(const ChainPosition &position) : in_(NULL) {
  Init(position);
}

Link::~Link() {
  if (current_) {
    // Probably an exception unwinding.
    std::cerr << "Last input should have been poison." << std::endl;
    abort();
  } else {
    if (!poisoned_) {
      // Poison is a block whose memory pointer is NULL.
      //
      // Because we're in the else block,
      //   we know that the memory pointer of current_ is NULL.
      //
      // Pass the current (poison) block!
      out_->Produce(current_);
    }
  }
}

Link &Link::operator++() {
  assert(current_);
  progress_ += current_.ValidSize();
  out_->Produce(current_);
  in_->Consume(current_);
  if (!current_) {
    poisoned_ = true;
    out_->Produce(current_);
  }
  return *this;
}

void Link::Poison() {
  assert(!poisoned_);
  current_.SetToPoison();
  out_->Produce(current_);
  poisoned_ = true;
}

} // namespace stream
} // namespace util

	#include "util/stream/chain.hh"

	#include "util/stream/io.hh"

	#include "util/exception.hh"
	#include "util/pcqueue.hh"

	#include <cstdlib>
	#include <new>
	#include <iostream>
	#include <stdint.h>

	namespace util {
	namespace stream {

	ChainConfigException::ChainConfigException() throw() { *this << "Chain configured with "; }
	ChainConfigException::~ChainConfigException() throw() {}

	Thread::~Thread() {
	thread_.join();
	}

	void Thread::UnhandledException(const std::exception &e) {
	std::cerr << e.what() << std::endl;
	abort();
	}

	void Recycler::Run(const ChainPosition &position) {
	for (Link l(position); l; ++l) {
	l->SetValidSize(position.GetChain().BlockSize());
	}
	}

	const Recycler kRecycle = Recycler();

	Chain::Chain(const ChainConfig &config) : config_(config), complete_called_(false) {
	UTIL_THROW_IF(!config.entry_size, ChainConfigException, "zero-size entries.");
	UTIL_THROW_IF(!config.block_count, ChainConfigException, "block count zero");
	UTIL_THROW_IF(config.total_memory < config.entry_size * config.block_count, ChainConfigException, config.total_memory << " total memory, too small for " << config.block_count << " blocks of containing entries of size " << config.entry_size);
	// Round down block size to a multiple of entry size.
	block_size_ = config.total_memory / (config.block_count * config.entry_size) * config.entry_size;
	}

	Chain::~Chain() {
	Wait();
	}

	ChainPosition Chain::Add() {
	if (!Running()) Start();
	PCQueue<Block> &in = queues_.back();
	queues_.push_back(new PCQueue<Block>(config_.block_count));
	return ChainPosition(in, queues_.back(), this, progress_);
	}

	Chain &Chain::operator>>(const WriteAndRecycle &writer) {
	threads_.push_back(new Thread(Complete(), writer));
	return *this;
	}

	Chain &Chain::operator>>(const PWriteAndRecycle &writer) {
	threads_.push_back(new Thread(Complete(), writer));
	return *this;
	}

	void Chain::Wait(bool release_memory) {
	if (queues_.empty()) {
	assert(threads_.empty());
	return; // Nothing to wait for.
	}
	if (!complete_called_) CompleteLoop();
	threads_.clear();
	for (std::size_t i = 0; queues_.front().Consume(); ++i) {
	if (i == config_.block_count) {
	std::cerr << "Chain ending without poison." << std::endl;
	abort();
	}
	}
	queues_.clear();
	progress_.Finished();
	complete_called_ = false;
	if (release_memory) memory_.reset();
	}

	void Chain::Start() {
	Wait(false);
	if (!memory_.get()) {
	// Allocate memory.
	assert(threads_.empty());
	assert(queues_.empty());
	std::size_t malloc_size = block_size_ * config_.block_count;
	memory_.reset(MallocOrThrow(malloc_size));
	}
	// This queue can accomodate all blocks.
	queues_.push_back(new PCQueue<Block>(config_.block_count));
	// Populate the lead queue with blocks.
	uint8_t base = static_cast<uint8_t>(memory_.get());
	for (std::size_t i = 0; i < config_.block_count; ++i) {
	queues_.front().Produce(Block(base, block_size_));
	base += block_size_;
	}
	}

	ChainPosition Chain::Complete() {
	assert(Running());
	UTIL_THROW_IF(complete_called_, util::Exception, "CompleteLoop() called twice");
	complete_called_ = true;
	return ChainPosition(queues_.back(), queues_.front(), this, progress_);
	}

	Link::Link() : in_(NULL), out_(NULL), poisoned_(true) {}

	void Link::Init(const ChainPosition &position) {
	UTIL_THROW_IF(in_, util::Exception, "Link::Init twice");
	in_ = position.in_;
	out_ = position.out_;
	poisoned_ = false;
	progress_ = position.progress_;
	in_->Consume(current_);
	}

	Link::Link(const ChainPosition &position) : in_(NULL) {
	Init(position);
	}

	Link::~Link() {
	if (current_) {
	// Probably an exception unwinding.
	std::cerr << "Last input should have been poison." << std::endl;
	abort();
	} else {
	if (!poisoned_) {
	// Poison is a block whose memory pointer is NULL.
	//
	// Because we're in the else block,
	// we know that the memory pointer of current_ is NULL.
	//
	// Pass the current (poison) block!
	out_->Produce(current_);
	}
	}
	}

	Link &Link::operator++() {
	assert(current_);
	progress_ += current_.ValidSize();
	out_->Produce(current_);
	in_->Consume(current_);
	if (!current_) {
	poisoned_ = true;
	out_->Produce(current_);
	}
	return *this;
	}

	void Link::Poison() {
	assert(!poisoned_);
	current_.SetToPoison();
	out_->Produce(current_);
	poisoned_ = true;
	}

	} // namespace stream
	} // namespace util