CaliperSensorGroup.cpp

//================================================================================
// Name        : CaliperSensorGroup.cpp
// Author      : Micha Mueller
// Contact     : info@dcdb.it
// Copyright   : Leibniz Supercomputing Centre
// Description : Source file for Caliper sensor group class.
//================================================================================

//================================================================================
// This file is part of DCDB (DataCenter DataBase)
// Copyright (C) 2019-2019 Leibniz Supercomputing Centre
//
// 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., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
//================================================================================

#include "CaliperSensorGroup.h"

#include <errno.h>
#include <stdio.h>

#include <sys/mman.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/un.h>

#include "timestamp.h"

CaliperSensorGroup::CaliperSensorGroup(const std::string& name) :
    SensorGroupTemplate(name),
    _socket(-1),
    _connection(-1),
    _shm(nullptr),
    _shmFile(-1),
    _shmFailCnt(0),
    _globalMqttPrefix("") {
    _lock.clear();
}

CaliperSensorGroup::CaliperSensorGroup(const CaliperSensorGroup& other) :
    SensorGroupTemplate(other),
    _socket(-1),
    _connection(-1),
    _shm(nullptr),
    _shmFile(-1),
    _shmFailCnt(0),
    _globalMqttPrefix(other._globalMqttPrefix) {
    _lock.clear();

    //SensorGroupTemplate already copy constructed _sensor
    for (auto& s : _sensors) {
        _sensorIndex.insert(std::make_pair(s->getName(), s));
    }
}

CaliperSensorGroup::~CaliperSensorGroup() {
    _sensorIndex.clear();
}

CaliperSensorGroup& CaliperSensorGroup::operator=(const CaliperSensorGroup& other) {
    SensorGroupTemplate::operator=(other);
    _socket = -1;
    _connection = -1;
    _globalMqttPrefix = other._globalMqttPrefix;
    _lock.clear();

    //SensorGroupTemplate already copied _sensor
    for (auto& s : _sensors) {
        _sensorIndex.insert(std::make_pair(s->getName(), s));
    }

    return *this;
}

bool CaliperSensorGroup::execOnStart() {
	_socket = socket(AF_UNIX, SOCK_SEQPACKET | SOCK_NONBLOCK, 0);

	if(_socket == -1) {
	    LOG(error) << _groupName << ": Failed to open socket: " << strerror(errno);
	    return false;
	}

	sockaddr_un addr;
	memset(&addr, 0, sizeof(struct sockaddr_un));

	addr.sun_family = AF_UNIX;
	snprintf(&addr.sun_path[1], 91, SOCK_NAME);

	if(bind(_socket, (struct sockaddr*) &addr, sizeof(addr))) {
	    LOG(error) << _groupName << ": Failed to bind socket: " << strerror(errno);
	    close(_socket);
	    _socket = -1;
	    return false;
	}

	if(listen(_socket, 1)) {
	    LOG(error) << _groupName << ": Can not listen on socket: " << strerror(errno);
	    close(_socket);
	    _socket = -1;
	    return false;
	}

	return true;
}

void CaliperSensorGroup::execOnStop() {
    if(_connection != -1) {
        close(_connection);
        _connection = -1;
    }
    if(_socket != -1) {
        close(_socket);
        _socket = -1;
    }
}

void CaliperSensorGroup::read() {
    if (_shm == nullptr) {
        //check if new application wants to send us its PID
        _connection = accept(_socket, NULL, NULL);
        if (_connection == -1) {
            if (errno != EAGAIN && errno != EWOULDBLOCK) {
                LOG(error) << _groupName << ": Accept failed: " << strerror(errno);
            }
            return;
        }
        //Clear sensors from last connection. If their values have not been pushed
        //by now they are lost.
        acquireSensors();
        _sensors.clear();
        _baseSensors.clear();
        releaseSensors();

        const size_t bufSize = 64;
        char buf[bufSize];

        const ssize_t nrec = recv(_connection, (void*) buf, bufSize, MSG_DONTWAIT);

        close(_connection);
        _connection = -1;

        if (nrec <= 0) {
            LOG(error) << _groupName << ": Connection accepted but got no message";
            return;
        }

        std::string pidStr(buf);

        _shmFile = shm_open((STR_PREFIX + pidStr).c_str(), O_RDWR, 0666);
        if (_shmFile == -1) {
            LOG(error) << _groupName << ": Failed to open _shmFile";
            return;
        }
        _shm = mmap(NULL, SHM_SIZE, PROT_READ | PROT_WRITE, MAP_SHARED, _shmFile, 0);
        if (_shm == (void*) -1) {
            LOG(error) << _groupName << ": Failed to mmap _shmFile";
            _shm = nullptr;
            close(_shmFile);
            _shmFile = -1;
            return;
        }
    }

    //get snapshot data from message queue in shared memory
    size_t r_index;
    size_t w_index;

    sem_t* r_sem;
    sem_t* w_sem;

    snap_data* msg_queue;

    r_sem = reinterpret_cast<sem_t*>(static_cast<char*>(_shm) + 2*sizeof(size_t));
    w_sem = r_sem + 1;
    msg_queue = reinterpret_cast<snap_data*>(w_sem + 1);

    //TODO atomic load/stores instead of semaphore locking?
    if (sem_wait(r_sem)) {
        return;
    }
    r_index = *(reinterpret_cast<size_t*>(static_cast<char*>(_shm)));
    sem_post(r_sem);

    if (sem_wait(w_sem)) {
        return;
    }
    w_index = *(reinterpret_cast<size_t*>(static_cast<char*>(_shm) + sizeof(size_t)));
    sem_post(w_sem);

    //are new elements there at all?
    if (r_index == w_index) {
        ++_shmFailCnt;
        if (_shmFailCnt > SHM_MAX_RETRIES) {
            //"Timeout". We assume that the application terminated
            _sensorIndex.clear();

            sem_destroy(r_sem);
            sem_destroy(w_sem);

            munmap(_shm, SHM_SIZE);
            _shm = nullptr;
            close(_shmFile);
            _shmFile = -1;
        }
        LOG(debug) << "No data available (failCnt=" << _shmFailCnt << ")";
        return;
    }

    _shmFailCnt = 0;
    size_t nelems = 0;
    snap_data snaps[MSGQ_SIZE];

    if (r_index < w_index) {
        nelems = w_index - r_index;
        memcpy(snaps, &msg_queue[r_index+1], nelems*sizeof(snap_data));
    } else {
        nelems = MSGQ_SIZE - r_index + w_index;
        size_t sep = MSGQ_SIZE - r_index - 1;
        memcpy(snaps, &msg_queue[r_index+1], sep*sizeof(snap_data));
        memcpy(&snaps[sep], msg_queue, (nelems-sep)*sizeof(snap_data));
    }

    //update r_index in _shm
    if (sem_wait(r_sem)) {
        return;
    }
    *(reinterpret_cast<size_t*>(static_cast<char*>(_shm))) = w_index;
    sem_post(r_sem);

    LOG(debug) << "Processing " << nelems << " snapshots";

    for (size_t i = 0; i < nelems; ++i) {
        reading_t reading;
        reading.value     = 1;
        reading.timestamp = snaps[i].ts;

        std::string sName("cpu" + std::to_string(snaps[i].cpu) + '/');
        uintptr_t pc = snaps[i].pc;

        const size_t addrCnt = *(reinterpret_cast<size_t*>(static_cast<char*>(_shm)
                                                            + lookup_data_offset));
        const addr_data* const addrs = reinterpret_cast<addr_data*>(static_cast<char*>(_shm)
                                                                    + lookup_data_offset + sizeof(size_t));

        for(size_t j = 0; j < addrCnt; ++j) {
            if (pc >= addrs[j].start_addr && pc <= addrs[j].end_addr) {
                sName += addrs[j].pathname;

                const fsym_data* const fsyms = reinterpret_cast<const fsym_data* const>(
                                               reinterpret_cast<const char* const>(addrs) + addrs[j].fsym_offset);
                for(size_t k = 0; k < addrs[j].fsym_count; ++k) {
                    if (pc >= fsyms[k].start_addr && pc <= fsyms[k].end_addr) {
                        sName += "::";
                        sName += fsyms[k].name;
                        break;
                    }
                } //It's OK if we found no symbol. There are possibly none

                //store in sensors
                //TODO aggregate values (see header file)
                S_Ptr s;
                auto it = _sensorIndex.find(sName);
                if(it != _sensorIndex.end()) {
                    //we encountered this function or event name already
                    s = it->second;
                } else {
                    //unknown function or event name --> create a new sensor
                    s = std::make_shared<CaliperSensorBase>(sName);
                    s->setMqtt(_globalMqttPrefix + _mqttPart + sName);
                    s->setName(s->getMqtt());
                    s->initSensor(_interval);

                    acquireSensors();
                    _sensors.push_back(s);
                    _baseSensors.push_back(s);
                    releaseSensors();
                    _sensorIndex.insert(std::make_pair(sName, s));
                }
                s->storeReading(reading);
#ifdef DEBUG
                LOG(debug) << _groupName << "::" << s->getName() << " raw reading: \"" << reading.value << "\"";
#endif
                break;
            }
        }
        //TODO what if pc was not within any range?
    }
}

void CaliperSensorGroup::printGroupConfig(LOG_LEVEL ll) {
    //nothing special to print
}