SlsDetectorCPUAffinity.cpp

//###########################################################################
// This file is part of LImA, a Library for Image Acquisition
//
// Copyright (C) : 2009-2011
// European Synchrotron Radiation Facility
// BP 220, Grenoble 38043
// FRANCE
//
// This 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 3 of the License, or
// (at your option) any later version.
//
// This software 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, see <http://www.gnu.org/licenses/>.
//###########################################################################

#include "SlsDetectorCamera.h"

#include "lima/Timestamp.h"
#include "lima/CtAcquisition.h"
#include "lima/CtSaving.h"
#include "lima/SoftOpExternalMgr.h"
#include "lima/RegExUtils.h"
#include "lima/MiscUtils.h"

#include <signal.h>
#include <unistd.h>
#include <sys/wait.h>
#include <sys/types.h>
#include <pwd.h>
#include <numa.h>
#include <iomanip>

using namespace std;
using namespace lima;
using namespace lima::SlsDetector;

bool SystemCmd::UseSudo = true;

SystemCmd::SystemCmd(string cmd, string desc, bool try_sudo)
	: m_cmd(cmd), m_desc(desc), m_try_sudo(try_sudo),
	  m_stdin(NULL), m_stdout(NULL), m_stderr(NULL)
{
	DEB_CONSTRUCTOR();
	DEB_PARAM() << DEB_VAR3(m_cmd, m_desc, m_try_sudo);
}

SystemCmd::SystemCmd(const SystemCmd& o)
	: m_cmd(o.m_cmd), m_desc(o.m_desc), m_try_sudo(o.m_try_sudo),
	  m_stdin(o.m_stdin), m_stdout(o.m_stdout), m_stderr(o.m_stderr)
{
	DEB_CONSTRUCTOR();
	DEB_PARAM() << DEB_VAR3(m_cmd, m_desc, m_try_sudo);
	DEB_PARAM() << DEB_VAR3(m_stdin, m_stdout, m_stderr);
}

void SystemCmd::setUseSudo(bool use_sudo)
{
	UseSudo = use_sudo;
}

bool SystemCmd::getUseSudo()
{
	return UseSudo;
}

void SystemCmd::checkSudo()
{
	DEB_STATIC_FUNCT();

	typedef map<string, bool> CacheMap;
	static CacheMap cache_map;
	CacheMap::iterator it = cache_map.find(m_cmd);
	if (it == cache_map.end()) {
		SystemCmd sudo("sudo", "", false);
		sudo.args() << "-l " << m_cmd;
		bool ok = (sudo.execute() == 0);
		CacheMap::value_type entry(m_cmd, ok);
		pair<CacheMap::iterator, bool> v = cache_map.insert(entry);
		if (!v.second)
			THROW_HW_ERROR(Error) << "Error inserting cache entry";
		it = v.first;
	}
	if (it->second)
		return;

	const char *user = "Unknown";
	uid_t uid = getuid();
	DEB_TRACE() << DEB_VAR1(uid);
	struct passwd pw, *res;
	char buffer[4 * 1024];
	int ret = getpwuid_r(uid, &pw, buffer, sizeof(buffer), &res);
	if (ret != 0)
		DEB_WARNING() << "getpwuid_r failed: " << strerror(errno);
	else if (res == NULL)
		DEB_WARNING() << "could not get passwd entry for uid=" << uid;
	else
		user = pw.pw_name;

	DEB_ERROR() << "The command '" << m_cmd << "' is not allowed for "
		    << user << " in the sudoers database. ";
	DEB_ERROR() << "Check sudoers(5) man page and restart this process";
	if (!m_desc.empty())
		DEB_ERROR() << m_desc;

	THROW_HW_ERROR(Error) << "Cannot execute sudo " << m_cmd << "! "
			      << "See output for details";
}

void SystemCmd::setPipes(Pipe *stdin, Pipe *stdout, Pipe *stderr)
{
	DEB_MEMBER_FUNCT();
	DEB_PARAM() << DEB_VAR3(stdin, stdout, stderr);
	m_stdin = stdin;
	m_stdout = stdout;
	m_stderr = stderr;
}

int SystemCmd::execute()
{
	DEB_MEMBER_FUNCT();
	string args = m_args.str();
	DEB_PARAM() << DEB_VAR3(m_cmd, args, m_try_sudo);

	ostringstream os;
	if (m_try_sudo && getUseSudo()) {
		checkSudo();
		os << "sudo -n ";
	}
	os << m_cmd << " " << args;
	if (!DEB_CHECK_ANY(DebTypeTrace) && !m_stdout)
		os << " > /dev/null";
	if (sameOutErr())
		os << " 2>&1";
	DEB_TRACE() << "executing: '" << os.str() << "'";
	preparePipes();
	int ret = system(os.str().c_str());
	restorePipes();
	DEB_RETURN() << DEB_VAR1(ret);
	return ret;
}

void SystemCmd::preparePipes()
{
	DEB_MEMBER_FUNCT();
	DEB_PARAM() << DEB_VAR3(m_stdin, m_stdout, m_stderr);
	if (m_stdin) {
		DEB_TRACE() << "Duplicating PIPE into STDIN";
		m_stdin->dupInto(Pipe::ReadFd, 0);
	}
	if (m_stderr && !sameOutErr()) {
		DEB_TRACE() << "Duplicating PIPE into STDERR";
		m_stderr->dupInto(Pipe::WriteFd, 2);
	}
	if (m_stdout) {
		DEB_TRACE() << "Duplicating PIPE into STDOUT";
		m_stdout->dupInto(Pipe::WriteFd, 1);
	}
}

void SystemCmd::restorePipes()
{
	DEB_MEMBER_FUNCT();
	DEB_PARAM() << DEB_VAR3(m_stdin, m_stdout, m_stderr);
	if (m_stdout) {
		m_stdout->restoreDup(Pipe::WriteFd);
		DEB_TRACE() << "Restored STDOUT";
	}
	if (m_stderr && !sameOutErr()) {
		m_stderr->restoreDup(Pipe::WriteFd);
		DEB_TRACE() << "Restored STDERR";
	}
	if (m_stdin) {
		DEB_TRACE() << "Restored STDIN";
		m_stdin->restoreDup(Pipe::ReadFd);
	}
}

SystemCmdPipe::SystemCmdPipe(string cmd, string desc, bool try_sudo)
	: m_pipe_list(NbPipes), m_child_pid(-1), m_cmd(cmd, desc, try_sudo)
{
	DEB_CONSTRUCTOR();
}

SystemCmdPipe::~SystemCmdPipe()
{
	DEB_DESTRUCTOR();

	if (m_child_pid >= 0)
		wait();
}

void SystemCmdPipe::start()
{
	DEB_MEMBER_FUNCT();

	if (m_child_pid >= 0)
		THROW_HW_ERROR(Error) << "cmd already running";

	m_cmd.setPipes(m_pipe_list[StdIn].ptr,
		       m_pipe_list[StdOut].ptr,
		       m_pipe_list[StdErr].ptr);

	m_child_pid = fork();
	if (m_child_pid == 0) {
		m_pipe_list[StdIn].close(Pipe::WriteFd);
		m_pipe_list[StdOut].close(Pipe::ReadFd);
		m_pipe_list[StdErr].close(Pipe::ReadFd);

		int ret = m_cmd.execute();
		DEB_RETURN() << DEB_VAR1(ret);
		_exit(ret);
	} else {
		m_pipe_list[StdIn].close(Pipe::ReadFd);
		m_pipe_list[StdOut].close(Pipe::WriteFd);
		m_pipe_list[StdErr].close(Pipe::WriteFd);
	}
}

void SystemCmdPipe::wait()
{
	DEB_MEMBER_FUNCT();
	if (m_child_pid < 0)
		THROW_HW_ERROR(Error) << "cmd not running";
	int child_ret;
	waitpid(m_child_pid, &child_ret, 0);
	DEB_TRACE() << DEB_VAR1(child_ret);
	m_child_pid = -1;
}

void SystemCmdPipe::setPipe(PipeIdx idx, PipeType type)
{
	DEB_MEMBER_FUNCT();
	m_pipe_list[idx] = PipeData(type);
}

Pipe& SystemCmdPipe::getPipe(PipeIdx idx)
{
	DEB_MEMBER_FUNCT();
	if (!m_pipe_list[idx])
		THROW_HW_ERROR(InvalidValue) << "null pipe " << DEB_VAR1(idx);
	return *m_pipe_list[idx].ptr;
}

int CPUAffinity::findNbSystemCPUs()
{
	DEB_STATIC_FUNCT();
	int nb_cpus = 0;
	const char *proc_file_name = "/proc/cpuinfo";
	ifstream proc_file(proc_file_name);
	while (proc_file) {
		char buffer[1024];
		proc_file.getline(buffer, sizeof(buffer));
		istringstream is(buffer);
		string t;
		is >> t;
		if (t == "processor")
			++nb_cpus;
	}
	DEB_RETURN() << DEB_VAR1(nb_cpus);
	return nb_cpus;
}

int CPUAffinity::findMaxNbSystemCPUs()
{
	DEB_STATIC_FUNCT();
	NumericGlob proc_glob("/proc/sys/kernel/sched_domain/cpu");
	int max_nb_cpus = proc_glob.getNbEntries();
	DEB_RETURN() << DEB_VAR1(max_nb_cpus);
	return max_nb_cpus;
}

int CPUAffinity::getNbSystemCPUs(bool max_nb)
{
	static int nb_cpus = 0;
	EXEC_ONCE(nb_cpus = findNbSystemCPUs());
	static int max_nb_cpus = 0;
	EXEC_ONCE(max_nb_cpus = findMaxNbSystemCPUs());
	int cpus = (max_nb && max_nb_cpus) ? max_nb_cpus : nb_cpus;
	return cpus;
}

std::string CPUAffinity::getProcDir(bool local_threads)
{
	DEB_STATIC_FUNCT();
	DEB_PARAM() << DEB_VAR1(local_threads);
	ostringstream os;
	os << "/proc/";
	if (local_threads)
		os << getpid() << "/task/";
	string proc_dir = os.str();
	DEB_RETURN() << DEB_VAR1(proc_dir);
	return proc_dir;
}

std::string CPUAffinity::getTaskProcDir(pid_t task, bool is_thread)
{
	DEB_STATIC_FUNCT();
	DEB_PARAM() << DEB_VAR2(task, is_thread);
	ostringstream os;
	os << getProcDir(is_thread) << task << "/";
	string proc_dir = os.str();
	DEB_RETURN() << DEB_VAR1(proc_dir);
	return proc_dir;
}

void CPUAffinity::initCPUSet(cpu_set_t& cpu_set) const
{
	CPU_ZERO(&cpu_set);
	uint64_t mask = getMask();
	for (unsigned int i = 0; i < sizeof(mask) * 8; ++i) {
		if ((mask >> i) & 1)
			CPU_SET(i, &cpu_set);
	}
}

void CPUAffinity::applyToTask(pid_t task, bool incl_threads,
			      bool use_taskset) const
{
	DEB_MEMBER_FUNCT();
	DEB_PARAM() << DEB_VAR4(*this, task, incl_threads, use_taskset);

	string proc_status = getTaskProcDir(task, !incl_threads) + "status";
	if (access(proc_status.c_str(), F_OK) != 0)
		return;

	if (use_taskset)
		applyWithTaskset(task, incl_threads);
	else
		applyWithSetAffinity(task, incl_threads);
}

void CPUAffinity::applyWithTaskset(pid_t task, bool incl_threads) const
{
	DEB_MEMBER_FUNCT();
	DEB_PARAM() << DEB_VAR3(*this, task, incl_threads);

	SystemCmd taskset("taskset");
	const char *all_tasks_opt = incl_threads ? "-a " : "";
	taskset.args() << all_tasks_opt << "-p " << *this << " " << task;
	if (taskset.execute() != 0) {
		const char *th = incl_threads ? "and threads " : "";
		THROW_HW_ERROR(Error) << "Error setting task " << task 
				      << " " << th << "CPU affinity";
	}
}

void CPUAffinity::applyWithSetAffinity(pid_t task, bool incl_threads) const
{
	DEB_MEMBER_FUNCT();
	DEB_PARAM() << DEB_VAR3(*this, task, incl_threads);

	IntList task_list;
	if (incl_threads) {
		string proc_dir = getTaskProcDir(task, false) + "task/";
		NumericGlob proc_glob(proc_dir);
		typedef NumericGlob::IntStringList IntStringList;
		IntStringList list = proc_glob.getIntPathList();
		if (list.empty())
			THROW_HW_ERROR(Error) << "Cannot find task " << task 
					      << " threads";
		IntStringList::const_iterator it, end = list.end();
		for (it = list.begin(); it != end; ++it)
			task_list.push_back(it->first);
	} else {
		task_list.push_back(task);
	}

	cpu_set_t cpu_set;
	initCPUSet(cpu_set);
	IntList::const_iterator it, end = task_list.end();
	for (it = task_list.begin(); it != end; ++it) {
		DEB_TRACE() << "setting " << task << " CPU mask: " << *this;
		int ret = sched_setaffinity(task, sizeof(cpu_set), &cpu_set);
		if (ret != 0) {
			const char *th = incl_threads ? "and threads " : "";
			THROW_HW_ERROR(Error) << "Error setting task " << task 
					      << " " << th << "CPU affinity";
		}
	}
}

void CPUAffinity::getNUMANodeMask(vector<unsigned long>& node_mask,
				  int& max_node)
{
	DEB_MEMBER_FUNCT();

	typedef vector<unsigned long> Array;

	int nb_nodes = numa_max_node() + 1;
	const int item_bits = sizeof(Array::reference) * 8;
	int nb_items = nb_nodes / item_bits;
	if (nb_nodes % item_bits != 0)
		++nb_items;

	DEB_PARAM() << DEB_VAR2(*this, nb_items);
	max_node = nb_nodes + 1;

	node_mask.assign(nb_items, 0);

	uint64_t mask = getMask();
	for (unsigned int i = 0; i < sizeof(mask) * 8; ++i) {
		if ((mask >> i) & 1) {
			unsigned int n = numa_node_of_cpu(i);
			Array::reference v = node_mask[n / item_bits];
			v |= 1L << (n % item_bits);
		}
	}

	if (DEB_CHECK_ANY(DebTypeReturn)) {
		ostringstream os;
		os << hex << "0x" << setw(nb_nodes / 4) << setfill('0');
		bool first = true;
		Array::reverse_iterator it, end = node_mask.rend();
		for (it = node_mask.rbegin(); it != end; ++it, first = false)
			os << (!first ? "," : "") << *it;
		DEB_RETURN() << "node_mask=" << os.str() << ", "
			     << DEB_VAR1(max_node);
	}
}

bool IrqMgr::m_irqbalance_stopped = false;
StringList IrqMgr::m_dev_list;

IrqMgr::IrqMgr(string net_dev)
{
	DEB_CONSTRUCTOR();
	DEB_PARAM() << DEB_VAR1(net_dev);

	if (!net_dev.empty())
		setDev(net_dev);
}

IrqMgr::~IrqMgr()
{
	DEB_DESTRUCTOR();
	DEB_PARAM() << DEB_VAR1(m_net_dev);

	if (isManaged(m_net_dev))
		updateRxQueueIrqAffinity(true);
}

void IrqMgr::setDev(string net_dev)
{
	DEB_CONSTRUCTOR();
	DEB_PARAM() << DEB_VAR1(net_dev);

	if (!m_net_dev.empty())
		THROW_HW_ERROR(Error) << "device already set";
	if (isManaged(net_dev))
		THROW_HW_ERROR(Error) << net_dev << " already managed";

	m_net_dev = net_dev;
}

bool IrqMgr::isManaged(string net_dev)
{
	if (net_dev.empty())
		return false;
	StringList::iterator it, end = m_dev_list.end();
	it = find(m_dev_list.begin(), end, net_dev);
	return (it != end);
}

IntList IrqMgr::getIrqList()
{
	DEB_MEMBER_FUNCT();
	DEB_PARAM() << DEB_VAR1(m_net_dev);

	string pci_irq_dir = (string("/sys/class/net/") + m_net_dev + 
			      "/device/msi_irqs/");
	NumericGlob pci_irqs(pci_irq_dir);
	typedef NumericGlob::IntStringList IntStringList;
	IntStringList dev_irqs = pci_irqs.getIntPathList();

	IntList act_irqs;
	ifstream proc_ints("/proc/interrupts");
	ostringstream os;
	int nb_cpus = CPUAffinity::getNbSystemCPUs();
	os << "[ \t]*"
	   << "(?P<irq>[0-9]+):[ \t]+"
	   << "(?P<counts>(([0-9]+)[ \t]+){" << nb_cpus << "})"
	   << "(?P<type>[-A-Za-z0-9_]+)[ \t]+"
	   << "(?P<name>[-A-Za-z0-9_]+)";
	DEB_TRACE() << DEB_VAR1(os.str());
	RegEx int_re(os.str());
	while (proc_ints) {
		char buffer[1024];
		proc_ints.getline(buffer, sizeof(buffer));
		string s = buffer;
		DEB_TRACE() << DEB_VAR1(s);
		RegEx::FullNameMatchType match;
		if (!int_re.matchName(s, match))
			continue;
		int irq;
		istringstream(match["irq"]) >> irq;
		DEB_TRACE() << "found match: " << irq << ": " 
			     << string(match["name"]);
		IntStringList::iterator it, end = dev_irqs.end();
		bool ok = false;
		for (it = dev_irqs.begin(); !ok && (it != end); ++it)
			ok = (it->first == irq);
		if (ok)
			act_irqs.push_back(irq);
	}

	DEB_RETURN() << PrettyIntList(act_irqs);
	return act_irqs;
}

void IrqMgr::updateRxQueueIrqAffinity(bool default_affinity)
{
	DEB_MEMBER_FUNCT();
	DEB_PARAM() << DEB_VAR2(m_net_dev, default_affinity);

	if (isManaged(m_net_dev) && default_affinity) {
		StringList::iterator it, end = m_dev_list.end();
		it = find(m_dev_list.begin(), end, m_net_dev);
		m_dev_list.erase(it);
		if (m_dev_list.empty())
			restoreIrqBalance();
	} else if (!isManaged(m_net_dev) && !default_affinity) {
		stopIrqBalance();
		m_dev_list.push_back(m_net_dev);
	}
}

void IrqMgr::stopIrqBalance()
{
	DEB_STATIC_FUNCT();
	DEB_PARAM() << DEB_VAR1(m_irqbalance_stopped);
	if (!getIrqBalanceActive()) {
		DEB_TRACE() << "nothing to do";
		return;
	} else if (m_irqbalance_stopped) {
		DEB_WARNING() << "irqbalance stopped and still running!";
	}
	setIrqBalanceActive(false);
	m_irqbalance_stopped = true;
}

void IrqMgr::restoreIrqBalance()
{
	DEB_STATIC_FUNCT();
	DEB_PARAM() << DEB_VAR1(m_irqbalance_stopped);
	if (!m_irqbalance_stopped)
		return;
	setIrqBalanceActive(true);
	m_irqbalance_stopped = false;
}

bool IrqMgr::getIrqBalanceActive()
{
	DEB_STATIC_FUNCT();
	SystemCmd bash("bash", "", false);
	ConstStr cmd = "ps -ef | grep -v grep | grep irqbalance";
	bash.args() << "-c '" << cmd << "'";
	bool act = (bash.execute() == 0);
	DEB_RETURN() << DEB_VAR1(act);
	return act;
}

void IrqMgr::setIrqBalanceActive(bool act)
{
	DEB_STATIC_FUNCT();
	DEB_PARAM() << DEB_VAR1(act);
	SystemCmd service("service");
	ConstStr cmd = act ? "start" : "stop";
	DEB_ALWAYS() << "irqbalance: executing " << cmd;
	service.args() << "irqbalance " << cmd;
	if (service.execute() != 0)
		THROW_HW_ERROR(Error) << "Could not " << cmd << " "
				      << "irqbalance service";
	DEB_ALWAYS() << "Done!";
}


NetDevRxQueueMgr::NetDevRxQueueMgr(string dev)
	: m_dev(dev), m_irq_mgr(dev)
{
	DEB_CONSTRUCTOR();
	DEB_PARAM() << DEB_VAR1(m_dev);
}

NetDevRxQueueMgr::~NetDevRxQueueMgr()
{
	DEB_DESTRUCTOR();
	if (!NetDevRxQueueAffinityMap_isDefault(m_aff_map))
		apply(NetDevRxQueueAffinityMap());
}

void NetDevRxQueueMgr::setDev(string dev)
{
	DEB_MEMBER_FUNCT();
	DEB_PARAM() << DEB_VAR1(dev);
	if (m_dev.empty()) {
		m_dev = dev;
		m_irq_mgr.setDev(dev);
	} else if (dev != m_dev) {
		THROW_HW_ERROR(InvalidValue) << "name mismatch: "
					     << DEB_VAR2(dev, m_dev);
	}
}

void NetDevRxQueueMgr::checkDev()
{
	DEB_MEMBER_FUNCT();
	if (m_dev.empty())
		THROW_HW_ERROR(InvalidValue) << "no device defined yet";
}

void NetDevRxQueueMgr::apply(int queue, const Affinity& queue_affinity)
{
	DEB_MEMBER_FUNCT();
	DEB_PARAM() << DEB_VAR3(m_dev, queue, queue_affinity);

	checkDev();
	apply(Irq, queue, queue_affinity.irq);	
	m_irq_mgr.updateRxQueueIrqAffinity(queue_affinity.irq.isDefault());
	apply(Processing, queue, queue_affinity.processing);
	m_aff_map[queue] = queue_affinity;
}

void NetDevRxQueueMgr::apply(const AffinityMap& affinity_map)
{
	m_aff_map.clear();
	if (NetDevRxQueueAffinityMap_isDefault(affinity_map)) {
		apply(-1, NetDevRxQueueCPUAffinity());
	} else {
		AffinityMap::const_iterator qit, qend = affinity_map.end();
		for (qit = affinity_map.begin(); qit != qend; ++qit)
			apply(qit->first, qit->second);
	}
}

void NetDevRxQueueMgr::apply(Task task, int queue, CPUAffinity a)
{
	DEB_MEMBER_FUNCT();
	DEB_PARAM() << DEB_VAR3(m_dev, queue, a);

	FileSetterData file_setter;
	if (task == Irq)
		getIrqFileSetterData(queue, file_setter);
	else
		getProcessingFileSetterData(queue, file_setter);

	static bool did_error_files[NbTasks] = {false, false};
	if (!did_error_files[task]) {
		const StringList& list = file_setter.file;
		StringList::const_iterator it, end = list.end();
		bool ok = true;
		for (it = list.begin(); ok && (it != end); ++it)
			ok = applyWithFile(*it, a);
		if (ok)
			return;
		DEB_WARNING() << "Could not write to files. Will try setter...";
		did_error_files[task] = true;
	}

	static bool did_error_setter[NbTasks] = {false, false};
	if (!did_error_setter[task]) {
		const StringList& list = file_setter.setter;
		StringList::const_iterator it, end = list.end();
		bool ok = true;
		for (it = list.begin(); ok && (it != end); ++it)
			ok = applyWithSetter(task, *it, a);
		if (ok)
			return;
		DEB_ERROR() << "Could not use setter";
		did_error_setter[task] = true;
	}
}

void NetDevRxQueueMgr::getIrqFileSetterData(int queue,
					    FileSetterData& file_setter)
{
	DEB_MEMBER_FUNCT();

	if (queue != -1)
		THROW_HW_ERROR(NotSupported) << "only all queues (-1) mode "
					     << "is supported";

	IntList act_irqs = m_irq_mgr.getIrqList();
	IntList::const_iterator it, end = act_irqs.end();
	for (it = act_irqs.begin(); it != end; ++it) {
		ostringstream os1, os2;
		os1 << "/proc/irq/" << *it << "/smp_affinity";
		file_setter.file.push_back(os1.str());
		os2 << *it;
		file_setter.setter.push_back(os2.str());
	}
}

void NetDevRxQueueMgr::getProcessingFileSetterData(int queue,
						   FileSetterData& file_setter)
{
	DEB_MEMBER_FUNCT();
	DEB_PARAM() << DEB_VAR1(m_dev);

	enum {
		File, Setter, NbLists,
	};

	string glob_str(string("/sys/class/net/" + m_dev + "/queues/rx-"));
	NumericGlob rps_cpus_glob(glob_str, "/rps_cpus");

	typedef NumericGlob::IntStringList IntStringList;
	IntStringList list_array[NbLists];
	list_array[File] = rps_cpus_glob.getIntPathList();
	list_array[Setter] = rps_cpus_glob.getIntSubPathList(6);
	IntList filter_list;
	if (queue == -2)
		filter_list = getRxQueueList();
	else if (queue != -1)
		filter_list.push_back(queue);
	for (int i = 0; !filter_list.empty() && (i < NbLists); ++i) {
		IntStringList& list = list_array[i];
		while (true) {
			IntStringList::iterator lit, lend = list.end();
			bool ok = true;
			for (lit = list.begin(); ok && (lit != lend); ++lit) {
				IntList::iterator fit, fend;
				fend = filter_list.end();
				fit = find(filter_list.begin(), fend,
					   lit->first);
				ok = (fit != fend);
				if (!ok)
					list.erase(lit);
			}
			if (ok)
				break;
		}
	}
	if (list_array[File].size() != list_array[Setter].size()) {
		THROW_HW_ERROR(Error) << "File and Setter lists differ";
	} else if (list_array[File].empty()) {
		DEB_WARNING() << "No Rx queue (" << queue << ") for " << m_dev;
		return;
	}

	StringList *file_setter_list[NbLists] = {&file_setter.file,
						 &file_setter.setter};
	for (int i = 0; i < NbLists; ++i) {
		const IntStringList& list = list_array[i];
		IntStringList::const_iterator it, end = list.end();
		for (it = list.begin(); it != end; ++it)
			file_setter_list[i]->push_back(it->second);
	}
}

bool NetDevRxQueueMgr::applyWithFile(const string& fname, CPUAffinity a)
{
	DEB_MEMBER_FUNCT();

	ostringstream os;
	os << hex << a.getZeroDefaultMask();
	DEB_TRACE() << "writing " << os.str() << " to " << fname;
	ofstream aff_file(fname.c_str());
	if (aff_file)
		aff_file << os.str();
	if (aff_file)
		aff_file.close();
	bool file_ok(aff_file);
	DEB_RETURN() << DEB_VAR1(file_ok);
	return file_ok;
}

bool NetDevRxQueueMgr::applyWithSetter(Task task, const string& irq_queue,
				       CPUAffinity a)
{
	DEB_MEMBER_FUNCT();
	DEB_PARAM() << DEB_VAR2(m_dev, irq_queue);

	static string desc = getSetterSudoDesc();
	SystemCmd setter(AffinitySetterName, desc);
	ConstStr task_opt = (task == Irq) ? "-i" : "-r";
	setter.args() << task_opt << " " << m_dev << " " << irq_queue << " "
		      << hex << "0x" << a.getZeroDefaultMask();
	bool setter_ok = (setter.execute() == 0);
	DEB_RETURN() << DEB_VAR1(setter_ok);
	return setter_ok;
}

string NetDevRxQueueMgr::getSetterSudoDesc()
{
	DEB_STATIC_FUNCT();

	const string& setter_name = AffinitySetterName;
	string dir = "/tmp";
	string fname = dir + "/" + setter_name + ".c";
	ofstream src_file(fname.c_str());
	if (src_file) {
		const StringList& SrcList = AffinitySetterSrc;
		StringList::const_iterator it, end = SrcList.end();
		for (it = SrcList.begin(); src_file && (it != end); ++it)
			src_file << *it << endl;
		if (src_file)
			src_file.close();
		if (!src_file)
			DEB_WARNING() << "Error writing to " << fname;
	} else {
		DEB_WARNING() << "Error creating " << fname;
	}

	ostringstream desc;
	string aux_setter = dir + "/" + setter_name;
	desc << "In order to create " << setter_name << ", compile " << fname
	     << " with the following commands: " << endl
	     << "  gcc -Wall -o " << aux_setter << " " << fname << endl
	     << "  su -c \"cp " << aux_setter << " /usr/local/bin\"" << endl;
	return desc.str();
}

const string NetDevRxQueueMgr::AffinitySetterName = 
					"netdev_set_queue_cpu_affinity";

static const char *NetDevRxQueueMgrAffinitySetterSrcCList[] = {
"#include <stdio.h>",
"#include <stdlib.h>",
"#include <string.h>",
"#include <errno.h>",
"#include <unistd.h>",
"#include <sys/types.h>",
"#include <sys/stat.h>",
"#include <fcntl.h>",
"",
"int main(int argc, char *argv[])",
"{",
"	char *dev, *irq_queue, *p, fname[256], buffer[128];",
"	int irq, rps, fd, len, ret;",
"	long aff;",
"",
"	if (argc != 5)",
"		exit(1);",
"	irq = (strcmp(argv[1], \"-i\") == 0);",
"	rps = (strcmp(argv[1], \"-r\") == 0);",
"	if (!irq && !rps)",
"		exit(2);",
"	if (!strlen(argv[2]) || !strlen(argv[3]) || !strlen(argv[4]))",
"		exit(2);",
"",
"	dev = argv[2];",
"	irq_queue = argv[3];",
"",
"	errno = 0;",
"	aff = strtol(argv[4], &p, 0);",
"	if (errno || *p)",
"		exit(3);",
"",
"	len = sizeof(fname);",
"	if (irq)",
"		ret = snprintf(fname, len, \"/proc/irq/%s/smp_affinity\",", 
"			       irq_queue);",
"	else",
"		ret = snprintf(fname, len, \"/sys/class/net/%s/queues/%s/rps_cpus\",", 
"			       dev, irq_queue);",
"	if ((ret < 0) || (ret == len))",
"		exit(4);",
"",
"	len = sizeof(buffer);",
"	ret = snprintf(buffer, len, \"%016lx\", aff);",
"	if ((ret < 0) || (ret == len))",
"		exit(5);",
"",
"	fd = open(fname, O_WRONLY);",
"	if (fd < 0)",
"		exit(6);",
"",	
"	for (p = buffer; *p; p += ret)",
"		if ((ret = write(fd, p, strlen(p))) < 0)",
"			exit(7);",
"",
"	if (close(fd) < 0)",
"		exit(8);",
"	return 0;",
"}",
};
const StringList NetDevRxQueueMgr::AffinitySetterSrc(
		C_LIST_ITERS(NetDevRxQueueMgrAffinitySetterSrcCList));

IntList NetDevRxQueueMgr::getRxQueueList()
{
	DEB_MEMBER_FUNCT();

	checkDev();

	IntList queue_list;

	SystemCmdPipe ethtool("ethtool");
	ethtool.args() << "-S " << m_dev;
	ethtool.setPipe(SystemCmdPipe::StdOut, SystemCmdPipe::DoPipe);
	ethtool.start();
	Pipe& child_out = ethtool.getPipe(SystemCmdPipe::StdOut);

	RegEx re("^[ \t]*rx_queue_(?P<queue>[0-9]+)_packets:[ \t]+"
		 "(?P<packets>[0-9]+)\n$");
	while (true) {
		string s = child_out.readLine(1024, "\n");
		if (s.empty())
			break;

		RegEx::FullNameMatchType match;
		if (!re.matchName(s, match))
			continue;

		int queue;
		istringstream(match["queue"]) >> queue;
		unsigned long packets;
		istringstream(match["packets"]) >> packets;
		if (packets != 0) {
			DEB_TRACE() << m_dev << " RxQueue " << queue << ": "
				    << packets << " packets";
			queue_list.push_back(queue);
		}
	}
	if (queue_list.empty())
		DEB_WARNING() << "No queue found for " << m_dev;

	DEB_RETURN() << DEB_VAR1(PrettyIntList(queue_list));
	return queue_list;
}

SystemCPUAffinityMgr::WatchDog::WatchDog()
	: m_cmd_pipe(0, true), m_res_pipe(0, true)
{
	DEB_CONSTRUCTOR();

	m_lima_pid = getpid();

	m_child_pid = fork();
	if (m_child_pid == 0) {
		m_child_pid = getpid();
		DEB_TRACE() << DEB_VAR2(m_lima_pid, m_child_pid);

		m_cmd_pipe.close(Pipe::WriteFd);
		m_res_pipe.close(Pipe::ReadFd);

		signal(SIGINT, SIG_IGN);
		signal(SIGTERM, sigTermHandler);

		childFunction();
		_exit(0);
	} else {
		m_cmd_pipe.close(Pipe::ReadFd);
		m_res_pipe.close(Pipe::WriteFd);

		sendChildCmd(Packet(Init));
		DEB_TRACE() << "Child is ready";
	}
}

SystemCPUAffinityMgr::WatchDog::~WatchDog()
{
	DEB_DESTRUCTOR();

	if (!childEnded()) {
		sendChildCmd(Packet(CleanUp));
		waitpid(m_child_pid, NULL, 0);
	}
}

void SystemCPUAffinityMgr::WatchDog::sigTermHandler(int /*signo*/)
{
}

bool SystemCPUAffinityMgr::WatchDog::childEnded()
{
	return (waitpid(m_child_pid, NULL, WNOHANG) != 0);
}

void SystemCPUAffinityMgr::WatchDog::sendChildCmd(const Packet& packet)
{
	DEB_MEMBER_FUNCT();
	DEB_PARAM() << DEB_VAR1(packet.cmd);

	if (childEnded())
		THROW_HW_ERROR(Error) << "Watchdog child process killed: " 
				      << m_child_pid;

	void *p = static_cast<void *>(const_cast<Packet *>(&packet));
	string s(static_cast<char *>(p), sizeof(packet));
	m_cmd_pipe.write(s);

	s = m_res_pipe.read(1);
	Cmd res = Cmd(s.data()[0]);
	if (res != Ok)
		THROW_HW_ERROR(Error) << "Invalid watchdog child ack";
	DEB_TRACE() << "Watchdog child acknowledged Ok";
}

SystemCPUAffinityMgr::WatchDog::Packet
SystemCPUAffinityMgr::WatchDog::readParentCmd()
{
	DEB_MEMBER_FUNCT();