aos/logging/binary_log_writer.cc - RealtimeRoboticsGroup/test - Gitiles

 #include <dirent.h>
 #include <errno.h>
 #include <fcntl.h>
 #include <mntent.h>
 #include <pwd.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 #include <sys/types.h>
 #include <time.h>
 #include <unistd.h>
 #include <string>

 #include <chrono>
 #include <map>
 #include <unordered_set>

 #include "aos/die.h"
 #include "aos/logging/binary_log_file.h"
 #include "aos/logging/implementations.h"
 #include "aos/queue_types.h"
 #include "aos/time/time.h"
 #include "aos/linux_code/configuration.h"
 #include "aos/linux_code/init.h"
 #include "aos/linux_code/ipc_lib/queue.h"

 namespace aos {
 namespace logging {
 namespace linux_code {
 namespace {

 void CheckTypeWritten(uint32_t type_id, LogFileWriter *writer,
                       ::std::unordered_set<uint32_t> *written_type_ids) {
   if (written_type_ids->count(type_id) > 0) return;
   if (MessageType::IsPrimitive(type_id)) return;

   const MessageType &type = type_cache::Get(type_id);
   for (int i = 0; i < type.number_fields; ++i) {
     CheckTypeWritten(type.fields[i]->type, writer, written_type_ids);
   }

   char buffer[1024];
   ssize_t size = type.Serialize(buffer, sizeof(buffer));
   if (size == -1) {
     LOG(WARNING, "%zu-byte buffer is too small to serialize type %s\n",
         sizeof(buffer), type.name.c_str());
     return;
   }
   LogFileMessageHeader *const output =
       writer->GetWritePosition(sizeof(LogFileMessageHeader) + size);

   output->time_sec = output->time_nsec = 0;
   output->source = getpid();
   output->name_size = 0;
   output->sequence = 0;
   output->level = FATAL;

   memcpy(output + 1, buffer, size);
   output->message_size = size;

   output->type = LogFileMessageHeader::MessageType::kStructType;
   futex_set(&output->marker);

   written_type_ids->insert(type_id);
 }

 void AllocateLogName(char **filename, const char *directory) {
   int fileindex = 0;
   DIR *const d = opendir(directory);
   if (d == nullptr) {
     PDie("could not open directory %s", directory);
   }
   int index = 0;
   while (true) {
     errno = 0;
     struct dirent *const dir = readdir(d);
     if (dir == nullptr) {
       if (errno == 0) {
         break;
       } else {
         PLOG(FATAL, "readdir(%p) failed", d);
       }
     } else {
       if (sscanf(dir->d_name, "aos_log-%d", &index) == 1) {
         if (index >= fileindex) {
           fileindex = index + 1;
         }
       }
     }
   }
   closedir(d);

   char previous[512];
   ::std::string path = ::std::string(directory) + "/aos_log-current";
   ssize_t len = ::readlink(path.c_str(), previous, sizeof(previous));
   if (len != -1) {
     previous[len] = '\0';
   } else {
     previous[0] = '\0';
     LOG(INFO, "Could not find aos_log-current\n");
     printf("Could not find aos_log-current\n");
   }
   if (asprintf(filename, "%s/aos_log-%03d", directory, fileindex) == -1) {
     PDie("couldn't create final name");
   }
   LOG(INFO, "Created log file (aos_log-%d) in directory (%s). Previous file "
             "was (%s).\n",
       fileindex, directory, previous);
   printf("Created log file (aos_log-%d) in directory (%s). Previous file was "
          "(%s).\n",
          fileindex, directory, previous);
 }

 #ifdef AOS_ARCHITECTURE_arm_frc
 bool FoundThumbDrive(const char *path) {
   FILE *mnt_fp = setmntent("/etc/mtab", "r");
   if (mnt_fp == nullptr) {
     Die("Could not open /etc/mtab");
   }

   bool found = false;
   struct mntent mntbuf;
   char buf[256];
   while (!found) {
     struct mntent *mount_list = getmntent_r(mnt_fp, &mntbuf, buf, sizeof(buf));
     if (mount_list == nullptr) {
       break;
     }
     if (strcmp(mount_list->mnt_dir, path) == 0) {
       found = true;
     }
   }
   endmntent(mnt_fp);
   return found;
 }

 bool FindDevice(char *device, size_t device_size) {
   char test_device[10];
   for (char i = 'a'; i < 'z'; ++i) {
     snprintf(test_device, sizeof(test_device), "/dev/sd%c", i);
     LOG(INFO, "Trying to access %s\n", test_device);
     if (access(test_device, F_OK) != -1) {
       snprintf(device, device_size, "sd%c", i);
       return true;
     }
   }
   return false;
 }
 #endif

 int BinaryLogReaderMain() {
   InitNRT();

 #ifdef AOS_ARCHITECTURE_arm_frc
   char folder[128];

   {
     char dev_name[8];
     while (!FindDevice(dev_name, sizeof(dev_name))) {
       LOG(INFO, "Waiting for a device\n");
       printf("Waiting for a device\n");
       sleep(5);
     }
     snprintf(folder, sizeof(folder), "/media/%s1", dev_name);
     while (!FoundThumbDrive(folder)) {
       LOG(INFO, "Waiting for %s\n", folder);
       printf("Waiting for %s\n", folder);
       sleep(1);
     }
     snprintf(folder, sizeof(folder), "/media/%s1/", dev_name);
   }

   if (access(folder, F_OK) == -1) {
 #else
   const char *folder = configuration::GetLoggingDirectory();
   if (access(folder, R_OK | W_OK) == -1) {
 #endif
     LOG(FATAL, "folder '%s' does not exist. please create it\n", folder);
   }
   LOG(INFO, "logging to folder '%s'\n", folder);

   char *tmp;
   AllocateLogName(&tmp, folder);
   char *tmp2;
   if (asprintf(&tmp2, "%s/aos_log-current", folder) == -1) {
     PLOG(WARNING, "couldn't create current symlink name");
   } else {
     if (unlink(tmp2) == -1 && (errno != EROFS && errno != ENOENT)) {
       LOG(WARNING, "unlink('%s') failed", tmp2);
     }
     if (symlink(tmp, tmp2) == -1) {
       PLOG(WARNING, "symlink('%s', '%s') failed", tmp, tmp2);
     }
     free(tmp2);
   }
   int fd = open(tmp, O_SYNC | O_APPEND | O_RDWR | O_CREAT,
                 S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
   free(tmp);
   if (fd == -1) {
     PLOG(FATAL, "opening file '%s' failed", tmp);
   }
   LogFileWriter writer(fd);

   RawQueue *queue = GetLoggingQueue();

   ::std::unordered_set<uint32_t> written_type_ids;
   off_t clear_type_ids_cookie = 0;

   while (true) {
     const LogMessage *const msg =
         static_cast<const LogMessage *>(queue->ReadMessage(RawQueue::kNonBlock));
     if (msg == NULL) {
       // If we've emptied the queue, then wait for a bit before starting to read
       // again so the queue can buffer up some logs. This avoids lots of context
       // switches and mutex contention which happens if we're constantly reading
       // new messages as they come in.
       ::std::this_thread::sleep_for(::std::chrono::milliseconds(100));
       continue;
     }

     const size_t raw_output_length =
         sizeof(LogFileMessageHeader) + msg->name_length + msg->message_length;
     size_t output_length = raw_output_length;
     if (msg->type == LogMessage::Type::kStruct) {
       output_length += sizeof(msg->structure.type_id) + sizeof(uint32_t) +
                        msg->structure.string_length;
       if (writer.ShouldClearSeekableData(&clear_type_ids_cookie,
                                          output_length)) {
         writer.ForceNewPage();
         written_type_ids.clear();
       }
       CheckTypeWritten(msg->structure.type_id, &writer, &written_type_ids);
     } else if (msg->type == LogMessage::Type::kMatrix) {
       output_length +=
           sizeof(msg->matrix.type) + sizeof(uint32_t) + sizeof(uint16_t) +
           sizeof(uint16_t) + msg->matrix.string_length;
       CHECK(MessageType::IsPrimitive(msg->matrix.type));
     }
     LogFileMessageHeader *const output = writer.GetWritePosition(output_length);
     char *output_strings = reinterpret_cast<char *>(output) + sizeof(*output);
     output->name_size = msg->name_length;
     output->message_size = msg->message_length;
     output->source = msg->source;
     output->level = msg->level;
     output->time_sec = msg->seconds;
     output->time_nsec = msg->nseconds;
     output->sequence = msg->sequence;
     memcpy(output_strings, msg->name, msg->name_length);

     switch (msg->type) {
       case LogMessage::Type::kString:
         memcpy(output_strings + msg->name_length, msg->message,
                msg->message_length);
         output->type = LogFileMessageHeader::MessageType::kString;
         break;
       case LogMessage::Type::kStruct: {
         char *position = output_strings + msg->name_length;

         memcpy(position, &msg->structure.type_id,
                sizeof(msg->structure.type_id));
         position += sizeof(msg->structure.type_id);
         output->message_size += sizeof(msg->structure.type_id);

         const uint32_t length = msg->structure.string_length;
         memcpy(position, &length, sizeof(length));
         position += sizeof(length);
         memcpy(position, msg->structure.serialized,
                length + msg->message_length);
         position += length + msg->message_length;
         output->message_size += sizeof(length) + length;

         output->type = LogFileMessageHeader::MessageType::kStruct;
       } break;
       case LogMessage::Type::kMatrix: {
         char *position = output_strings + msg->name_length;

         memcpy(position, &msg->matrix.type, sizeof(msg->matrix.type));
         position += sizeof(msg->matrix.type);
         output->message_size += sizeof(msg->matrix.type);

         uint32_t length = msg->matrix.string_length;
         memcpy(position, &length, sizeof(length));
         position += sizeof(length);
         output->message_size += sizeof(length);

         uint16_t rows = msg->matrix.rows, cols = msg->matrix.cols;
         memcpy(position, &rows, sizeof(rows));
         position += sizeof(rows);
         memcpy(position, &cols, sizeof(cols));
         position += sizeof(cols);
         output->message_size += sizeof(rows) + sizeof(cols);
         CHECK_EQ(msg->message_length,
                  MessageType::Sizeof(msg->matrix.type) * rows * cols);

         memcpy(position, msg->matrix.data, msg->message_length + length);
         output->message_size += length;

         output->type = LogFileMessageHeader::MessageType::kMatrix;
       } break;
     }

     if (output->message_size - msg->message_length !=
         output_length - raw_output_length) {
       LOG(FATAL, "%zu != %zu\n", output->message_size - msg->message_length,
           output_length - raw_output_length);
     }

     futex_set(&output->marker);

     queue->FreeMessage(msg);
   }

   Cleanup();
   return 0;
 }

 }  // namespace
 }  // namespace linux_code
 }  // namespace logging
 }  // namespace aos

 int main() {
   return ::aos::logging::linux_code::BinaryLogReaderMain();
 }
	#include <dirent.h>
	#include <errno.h>
	#include <fcntl.h>
	#include <mntent.h>
	#include <pwd.h>
	#include <stdio.h>
	#include <stdlib.h>
	#include <string.h>
	#include <sys/types.h>
	#include <time.h>
	#include <unistd.h>
	#include <string>

	#include <chrono>
	#include <map>
	#include <unordered_set>

	#include "aos/die.h"
	#include "aos/logging/binary_log_file.h"
	#include "aos/logging/implementations.h"
	#include "aos/queue_types.h"
	#include "aos/time/time.h"
	#include "aos/linux_code/configuration.h"
	#include "aos/linux_code/init.h"
	#include "aos/linux_code/ipc_lib/queue.h"

	namespace aos {
	namespace logging {
	namespace linux_code {
	namespace {

	void CheckTypeWritten(uint32_t type_id, LogFileWriter *writer,
	::std::unordered_set<uint32_t> *written_type_ids) {
	if (written_type_ids->count(type_id) > 0) return;
	if (MessageType::IsPrimitive(type_id)) return;

	const MessageType &type = type_cache::Get(type_id);
	for (int i = 0; i < type.number_fields; ++i) {
	CheckTypeWritten(type.fields[i]->type, writer, written_type_ids);
	}

	char buffer[1024];
	ssize_t size = type.Serialize(buffer, sizeof(buffer));
	if (size == -1) {
	LOG(WARNING, "%zu-byte buffer is too small to serialize type %s\n",
	sizeof(buffer), type.name.c_str());
	return;
	}
	LogFileMessageHeader *const output =
	writer->GetWritePosition(sizeof(LogFileMessageHeader) + size);

	output->time_sec = output->time_nsec = 0;
	output->source = getpid();
	output->name_size = 0;
	output->sequence = 0;
	output->level = FATAL;

	memcpy(output + 1, buffer, size);
	output->message_size = size;

	output->type = LogFileMessageHeader::MessageType::kStructType;
	futex_set(&output->marker);

	written_type_ids->insert(type_id);
	}

	void AllocateLogName(char *filename, const char directory) {
	int fileindex = 0;
	DIR *const d = opendir(directory);
	if (d == nullptr) {
	PDie("could not open directory %s", directory);
	}
	int index = 0;
	while (true) {
	errno = 0;
	struct dirent *const dir = readdir(d);
	if (dir == nullptr) {
	if (errno == 0) {
	break;
	} else {
	PLOG(FATAL, "readdir(%p) failed", d);
	}
	} else {
	if (sscanf(dir->d_name, "aos_log-%d", &index) == 1) {
	if (index >= fileindex) {
	fileindex = index + 1;
	}
	}
	}
	}
	closedir(d);

	char previous[512];
	::std::string path = ::std::string(directory) + "/aos_log-current";
	ssize_t len = ::readlink(path.c_str(), previous, sizeof(previous));
	if (len != -1) {
	previous[len] = '\0';
	} else {
	previous[0] = '\0';
	LOG(INFO, "Could not find aos_log-current\n");
	printf("Could not find aos_log-current\n");
	}
	if (asprintf(filename, "%s/aos_log-%03d", directory, fileindex) == -1) {
	PDie("couldn't create final name");
	}
	LOG(INFO, "Created log file (aos_log-%d) in directory (%s). Previous file "
	"was (%s).\n",
	fileindex, directory, previous);
	printf("Created log file (aos_log-%d) in directory (%s). Previous file was "
	"(%s).\n",
	fileindex, directory, previous);
	}

	#ifdef AOS_ARCHITECTURE_arm_frc
	bool FoundThumbDrive(const char *path) {
	FILE *mnt_fp = setmntent("/etc/mtab", "r");
	if (mnt_fp == nullptr) {
	Die("Could not open /etc/mtab");
	}

	bool found = false;
	struct mntent mntbuf;
	char buf[256];
	while (!found) {
	struct mntent *mount_list = getmntent_r(mnt_fp, &mntbuf, buf, sizeof(buf));
	if (mount_list == nullptr) {
	break;
	}
	if (strcmp(mount_list->mnt_dir, path) == 0) {
	found = true;
	}
	}
	endmntent(mnt_fp);
	return found;
	}

	bool FindDevice(char *device, size_t device_size) {
	char test_device[10];
	for (char i = 'a'; i < 'z'; ++i) {
	snprintf(test_device, sizeof(test_device), "/dev/sd%c", i);
	LOG(INFO, "Trying to access %s\n", test_device);
	if (access(test_device, F_OK) != -1) {
	snprintf(device, device_size, "sd%c", i);
	return true;
	}
	}
	return false;
	}
	#endif

	int BinaryLogReaderMain() {
	InitNRT();

	#ifdef AOS_ARCHITECTURE_arm_frc
	char folder[128];

	{
	char dev_name[8];
	while (!FindDevice(dev_name, sizeof(dev_name))) {
	LOG(INFO, "Waiting for a device\n");
	printf("Waiting for a device\n");
	sleep(5);
	}
	snprintf(folder, sizeof(folder), "/media/%s1", dev_name);
	while (!FoundThumbDrive(folder)) {
	LOG(INFO, "Waiting for %s\n", folder);
	printf("Waiting for %s\n", folder);
	sleep(1);
	}
	snprintf(folder, sizeof(folder), "/media/%s1/", dev_name);
	}

	if (access(folder, F_OK) == -1) {
	#else
	const char *folder = configuration::GetLoggingDirectory();
	if (access(folder, R_OK \| W_OK) == -1) {
	#endif
	LOG(FATAL, "folder '%s' does not exist. please create it\n", folder);
	}
	LOG(INFO, "logging to folder '%s'\n", folder);

	char *tmp;
	AllocateLogName(&tmp, folder);
	char *tmp2;
	if (asprintf(&tmp2, "%s/aos_log-current", folder) == -1) {
	PLOG(WARNING, "couldn't create current symlink name");
	} else {
	if (unlink(tmp2) == -1 && (errno != EROFS && errno != ENOENT)) {
	LOG(WARNING, "unlink('%s') failed", tmp2);
	}
	if (symlink(tmp, tmp2) == -1) {
	PLOG(WARNING, "symlink('%s', '%s') failed", tmp, tmp2);
	}
	free(tmp2);
	}
	int fd = open(tmp, O_SYNC \| O_APPEND \| O_RDWR \| O_CREAT,
	S_IRUSR \| S_IWUSR \| S_IRGRP \| S_IWGRP \| S_IROTH \| S_IWOTH);
	free(tmp);
	if (fd == -1) {
	PLOG(FATAL, "opening file '%s' failed", tmp);
	}
	LogFileWriter writer(fd);

	RawQueue *queue = GetLoggingQueue();

	::std::unordered_set<uint32_t> written_type_ids;
	off_t clear_type_ids_cookie = 0;

	while (true) {
	const LogMessage *const msg =
	static_cast<const LogMessage *>(queue->ReadMessage(RawQueue::kNonBlock));
	if (msg == NULL) {
	// If we've emptied the queue, then wait for a bit before starting to read
	// again so the queue can buffer up some logs. This avoids lots of context
	// switches and mutex contention which happens if we're constantly reading
	// new messages as they come in.
	::std::this_thread::sleep_for(::std::chrono::milliseconds(100));
	continue;
	}

	const size_t raw_output_length =
	sizeof(LogFileMessageHeader) + msg->name_length + msg->message_length;
	size_t output_length = raw_output_length;
	if (msg->type == LogMessage::Type::kStruct) {
	output_length += sizeof(msg->structure.type_id) + sizeof(uint32_t) +
	msg->structure.string_length;
	if (writer.ShouldClearSeekableData(&clear_type_ids_cookie,
	output_length)) {
	writer.ForceNewPage();
	written_type_ids.clear();
	}
	CheckTypeWritten(msg->structure.type_id, &writer, &written_type_ids);
	} else if (msg->type == LogMessage::Type::kMatrix) {
	output_length +=
	sizeof(msg->matrix.type) + sizeof(uint32_t) + sizeof(uint16_t) +
	sizeof(uint16_t) + msg->matrix.string_length;
	CHECK(MessageType::IsPrimitive(msg->matrix.type));
	}
	LogFileMessageHeader *const output = writer.GetWritePosition(output_length);
	char output_strings = reinterpret_cast<char >(output) + sizeof(*output);
	output->name_size = msg->name_length;
	output->message_size = msg->message_length;
	output->source = msg->source;
	output->level = msg->level;
	output->time_sec = msg->seconds;
	output->time_nsec = msg->nseconds;
	output->sequence = msg->sequence;
	memcpy(output_strings, msg->name, msg->name_length);

	switch (msg->type) {
	case LogMessage::Type::kString:
	memcpy(output_strings + msg->name_length, msg->message,
	msg->message_length);
	output->type = LogFileMessageHeader::MessageType::kString;
	break;
	case LogMessage::Type::kStruct: {
	char *position = output_strings + msg->name_length;

	memcpy(position, &msg->structure.type_id,
	sizeof(msg->structure.type_id));
	position += sizeof(msg->structure.type_id);
	output->message_size += sizeof(msg->structure.type_id);

	const uint32_t length = msg->structure.string_length;
	memcpy(position, &length, sizeof(length));
	position += sizeof(length);
	memcpy(position, msg->structure.serialized,
	length + msg->message_length);
	position += length + msg->message_length;
	output->message_size += sizeof(length) + length;

	output->type = LogFileMessageHeader::MessageType::kStruct;
	} break;
	case LogMessage::Type::kMatrix: {
	char *position = output_strings + msg->name_length;

	memcpy(position, &msg->matrix.type, sizeof(msg->matrix.type));
	position += sizeof(msg->matrix.type);
	output->message_size += sizeof(msg->matrix.type);

	uint32_t length = msg->matrix.string_length;
	memcpy(position, &length, sizeof(length));
	position += sizeof(length);
	output->message_size += sizeof(length);

	uint16_t rows = msg->matrix.rows, cols = msg->matrix.cols;
	memcpy(position, &rows, sizeof(rows));
	position += sizeof(rows);
	memcpy(position, &cols, sizeof(cols));
	position += sizeof(cols);
	output->message_size += sizeof(rows) + sizeof(cols);
	CHECK_EQ(msg->message_length,
	MessageType::Sizeof(msg->matrix.type) * rows * cols);

	memcpy(position, msg->matrix.data, msg->message_length + length);
	output->message_size += length;

	output->type = LogFileMessageHeader::MessageType::kMatrix;
	} break;
	}

	if (output->message_size - msg->message_length !=
	output_length - raw_output_length) {
	LOG(FATAL, "%zu != %zu\n", output->message_size - msg->message_length,
	output_length - raw_output_length);
	}

	futex_set(&output->marker);

	queue->FreeMessage(msg);
	}

	Cleanup();
	return 0;
	}

	} // namespace
	} // namespace linux_code
	} // namespace logging
	} // namespace aos

	int main() {
	return ::aos::logging::linux_code::BinaryLogReaderMain();
	}