src/monitor/SHMReader.cpp

   1 /** Aesalon, a tool to visualize program behaviour in real time.
   2         Copyright (C) 2009-2011, Aesalon development team.
   3
   4         Aesalon is distributed under the terms of the GNU GPLv3. See
   5         the included file LICENSE for more information.
   6
   7         @file src/monitor/SHMReader.cpp
   8 */
   9
  10 #include <semaphore.h>
  11 #include <sys/mman.h>
  12 #include <fcntl.h>
  13 #include <cstring>
  14 #include <errno.h>
  15
  16 #include "Config.h"
  17
  18 #include "monitor/SHMReader.h"
  19 #include "monitor/Coordinator.h"
  20 #include "util/StreamAsString.h"
  21 #include "util/MessageSystem.h"
  22 #include "util/StringTo.h"
  23 #include "shm/ZoneHeader.h"
  24
  25 namespace Monitor {
  26
  27 SHMReader::ReadBroker::ReadBroker() : m_temporaryBuffer(NULL), m_temporaryBufferSize(0) {
  28
  29 }
  30
  31 SHMReader::ReadBroker::~ReadBroker() {
  32         if(m_temporaryBuffer) delete[] m_temporaryBuffer;
  33 }
  34
  35 void SHMReader::ReadBroker::resizeBuffer(uint32_t newSize) {
  36         if(m_temporaryBufferSize >= newSize) return;
  37         delete[] m_temporaryBuffer;
  38         m_temporaryBuffer = new uint8_t[newSize];
  39 }
  40
  41 SHMReader::SHMReader() : m_header(NULL) {
  42         m_shmName = Util::StreamAsString() << "/Aesalon-" << getpid();
  43
  44         m_fd = shm_open(m_shmName.c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR);
  45         if(m_fd == -1) Message(Fatal, "Could not open shared memory: " << strerror(errno));
  46
  47         setupHeader();
  48         setupConfiguration();
  49         setupZones();
  50 }
  51
  52 SHMReader::~SHMReader() {
  53
  54 }
  55
  56 uint32_t SHMReader::zoneCount() {
  57         return m_header->zoneCount;
  58 }
  59
  60 int32_t SHMReader::zoneWithData() {
  61         for(uint32_t i = 0; i < m_header->zonesAllocated; i ++) {
  62                 if(m_zoneUseData[i % 8] & (0x01 << (i % 8))) {
  63                         uint8_t *zone = getZone(i);
  64
  65                         if(zone == NULL) {
  66                                 Message(Warning, "Could not open zone #" << i);
  67                                 continue;
  68                         }
  69
  70                         SHM::ZoneHeader *zheader = reinterpret_cast<SHM::ZoneHeader *>(zone);
  71
  72                         if(sem_trywait(&zheader->packetSemaphore) == -1 && errno == EAGAIN) continue;
  73
  74                         return i;
  75                 }
  76         }
  77         return -1;
  78 }
  79
  80 uint32_t SHMReader::zoneProcessID(uint32_t zoneID) {
  81         SHM::ZoneHeader *header = reinterpret_cast<SHM::ZoneHeader *>(getZone(zoneID));
  82         return header->processID;
  83 }
  84
  85 uint32_t SHMReader::zoneThreadID(uint32_t zoneID) {
  86         SHM::ZoneHeader *header = reinterpret_cast<SHM::ZoneHeader *>(getZone(zoneID));
  87         return header->threadID;
  88 }
  89
  90 void SHMReader::waitForPacket() {
  91         sem_wait(&m_header->packetSemaphore);
  92 }
  93
  94 void SHMReader::processRequest(ReadBroker &request) {
  95         uint8_t *zone = getZone(request.zoneID());
  96
  97         SHM::ZoneHeader *header = reinterpret_cast<SHM::ZoneHeader *>(zone);
  98
  99         uint32_t start = header->head;
 100         uint32_t size1 = std::min(
 101                 request.size(),
 102                 (m_header->zoneSize*AesalonPageSize - header->gapSize) - start);
 103
 104         uint32_t size2 = request.size() - size1;
 105
 106         /* The difficult case; the data is in two segments. */
 107         if(size2 > 0) {
 108                 request.resizeBuffer(request.size());
 109                 memcpy(request.temporaryBuffer(), zone + start, size1);
 110                 memcpy(request.temporaryBuffer() + size1, zone + ZoneDataOffset, size2);
 111                 request.setData(request.temporaryBuffer());
 112                 header->head = size2 + ZoneDataOffset;
 113                 if(header->overflow > 0) {
 114                         Message(Debug, "overflow > 0 in difficult case, reducing . . .");
 115                         header->overflow -= size1;
 116                         header->overflow -= header->gapSize;
 117                         header->overflow -= size2;
 118                         header->gapSize = 0;
 119                         if(header->overflow <= 0) {
 120                                 Message(Debug, "Posting to overflow semaphore . . .");
 121                                 sem_post(&header->overflowSemaphore);
 122                         }
 123                 }
 124                 else header->gapSize = 0;
 125         }
 126         /* Otherwise, the simple case. */
 127         else {
 128                 request.setData(zone + start);
 129                 header->head += size1;
 130                 if(header->overflow > 0) {
 131                         Message(Debug, "overflow > 0, reducing . . .");
 132                         header->overflow -= size1;
 133                         Message(Debug, "New overflow: " << header->overflow);
 134                         if(header->overflow <= 0) {
 135                                 Message(Debug, "Posting to overflow semaphore . . .");
 136                                 sem_post(&header->overflowSemaphore);
 137                         }
 138                 }
 139         }
 140 }
 141
 142 uint8_t *SHMReader::getZone(uint32_t id) {
 143         if(id < m_zoneList.size() && m_zoneList[id] != NULL) {
 144                 return m_zoneList[id];
 145         }
 146
 147         uint8_t *zone =
 148                 static_cast<uint8_t *>(mapRegion(m_header->zonePageOffset + id*m_header->zoneSize, m_header->zoneSize));
 149
 150         m_zoneList.resize(id+1);
 151         m_zoneList[id] = zone;
 152
 153         return zone;
 154 }
 155
 156 void *SHMReader::mapRegion(uint32_t start, uint32_t size) {
 157         if(m_header == NULL || m_header->shmSize < (start+size)) {
 158                 if(m_header) sem_wait(&m_header->resizeSemaphore);
 159
 160                 Message(Debug, "Resizing SHM to " << start+size << " page(s).");
 161                 if(ftruncate(m_fd, (start+size) * AesalonPageSize) != 0) {
 162                         Message(Fatal, "Could not resize shared memory.");
 163                 }
 164                 if(m_header) m_header->shmSize = start+size;
 165
 166                 if(m_header) sem_post(&m_header->resizeSemaphore);
 167         }
 168
 169         void *memory =
 170                 mmap(NULL, size * AesalonPageSize, PROT_READ | PROT_WRITE, MAP_SHARED, m_fd, start * AesalonPageSize);
 171
 172         return memory;
 173 }
 174
 175 void SHMReader::unmapRegion(void *data, uint32_t size) {
 176         munmap(data, size*AesalonPageSize);
 177 }
 178
 179 void SHMReader::setupHeader() {
 180         m_header = reinterpret_cast<SHM::Header *>(mapRegion(0, 1));
 181
 182         sem_init(&m_header->packetSemaphore, 1, 0);
 183         sem_init(&m_header->resizeSemaphore, 1, 1);
 184 }
 185
 186 void SHMReader::setupConfiguration() {
 187         char *configurationData = static_cast<char *>(mapRegion(1, 1));
 188         m_header->configDataSize = 1;
 189
 190         std::vector<Config::Vault::KeyPair> configItems;
 191         Coordinator::instance()->vault()->match("*", configItems);
 192
 193         uint32_t offset = 0;
 194
 195         for(std::vector<Config::Vault::KeyPair>::iterator i = configItems.begin(); i != configItems.end(); ++i) {
 196                 /* Ignore all internal items. */
 197                 if(i->first.find("::") == 0) continue;
 198
 199                 while((offset + i->first.length() + i->second.length() + 2) > m_header->configDataSize*AesalonPageSize) {
 200                         unmapRegion(configurationData, m_header->configDataSize);
 201                         m_header->configDataSize ++;
 202                         configurationData = static_cast<char *>(mapRegion(1, m_header->configDataSize));
 203                 }
 204                 memcpy(&configurationData[offset], i->first.c_str(), i->first.length()+1);
 205                 offset += i->first.length()+1;
 206                 memcpy(&configurationData[offset], i->second.c_str(), i->second.length()+1);
 207                 offset += i->second.length()+1;
 208         }
 209 }
 210
 211 void SHMReader::setupZones() {
 212         m_header->zoneCount = 0;
 213         m_header->zoneSize = Util::StringTo<uint32_t>(Coordinator::instance()->vault()->get("zoneSize"));
 214         m_header->zoneUsagePages = Util::StringTo<uint32_t>(Coordinator::instance()->vault()->get("zoneUsePages"));
 215
 216         m_header->zonePageOffset = m_header->configDataSize + 1 + m_header->zoneUsagePages;
 217
 218         m_zoneUseData = static_cast<uint8_t *>(mapRegion(m_header->configDataSize + 1, m_header->zoneUsagePages));
 219 }
 220
 221 } // namespace Monitor