hydrar/mistserver
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Remove Windows-specific shared memory implementation

Browse source
This commit is contained in:
Thulinma 2024-05-02 16:01:05 +02:00
parent e75381ae39
commit 3422b1dbcd
6 changed files with 24 additions and 234 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

42
lib/defines.h
View file

@ -153,11 +153,11 @@ static inline void show_stackframe(){}
#define FLIP_MIN_DURATION 20000
// New meta
#define SHM_STREAM_META "MstMeta%s" //%s stream name
#define SHM_STREAM_META "/MstMeta%s" //%s stream name
#define SHM_STREAM_META_LEN 8 * 1024 * 1024
#define SHM_STREAM_META_ITEM 2 * 1024
#define SHM_STREAM_TM "MstTrak%s@%" PRIu32 "-%zu" //%s stream name
#define SHM_STREAM_TM "/MstTrak%s@%" PRIu32 "-%zu" //%s stream name
#define SHM_STREAM_TRACK_ITEM 16 * 1024 * 1024
#define SHM_STREAM_TRACK_LEN 4 * SHM_STREAM_TRACK_ITEM
@ -195,34 +195,34 @@ static inline void show_stackframe(){}
#define TRACK_PAGE_OFFSET 100
#define TRACK_PAGE_RECORDSIZE 36
#define COMMS_STATISTICS "MstStat"
#define COMMS_STATISTICS "/MstStat"
#define COMMS_STATISTICS_INITSIZE 16 * 1024 * 1024
#define COMMS_USERS "MstUser%s" //%s stream name
#define COMMS_USERS "/MstUser%s" //%s stream name
#define COMMS_USERS_INITSIZE 512 * 1024
#define COMMS_SESSIONS "MstSession%s"
#define COMMS_SESSIONS "/MstSession%s"
#define COMMS_SESSIONS_INITSIZE 8 * 1024 * 1024
#define CUSTOM_VARIABLES_INITSIZE 64 * 1024
#define EXTWRITERS "MstExtWriters"
#define EXTWRITERS "/MstExtWriters"
#define EXTWRITERS_INITSIZE 1 * 1024 * 1024
#define SEM_STATISTICS "/MstStat"
#define SEM_USERS "/MstUser%s" //%s stream name
#define SHM_TRACK_DATA "MstData%s@%zu_%" PRIu32 //%s stream name, %zu track ID, %PRIu32 page #
#define SHM_TRACK_DATA "/MstData%s@%zu_%" PRIu32 //%s stream name, %zu track ID, %PRIu32 page #
// End new meta
#define INPUT_USER_INTERVAL 250
#define SHM_STREAM_STATE "MstSTATE%s" //%s stream name
#define SHM_STREAM_CONF "MstSCnf%s" //%s stream name
#define SHM_STREAM_IPID "MstIPID%s" //%s stream name
#define SHM_STREAM_PPID "MstPPID%s" //%s stream name
#define SHM_GLOBAL_CONF "MstGlobalConfig"
#define SHM_STREAM_STATE "/MstSTATE%s" //%s stream name
#define SHM_STREAM_CONF "/MstSCnf%s" //%s stream name
#define SHM_STREAM_IPID "/MstIPID%s" //%s stream name
#define SHM_STREAM_PPID "/MstPPID%s" //%s stream name
#define SHM_GLOBAL_CONF "/MstGlobalConfig"
#define STRMSTAT_OFF 0
#define STRMSTAT_INIT 1
#define STRMSTAT_BOOT 2
@ -231,20 +231,20 @@ static inline void show_stackframe(){}
#define STRMSTAT_SHUTDOWN 5
#define STRMSTAT_INVALID 255
#define SHM_TRIGGER "MstTRGR%s" //%s trigger name
#define SHM_TRIGGER "/MstTRGR%s" //%s trigger name
#define SEM_LIVE "/MstLIVE%s" //%s stream name
#define SEM_INPUT "/MstInpt%s" //%s stream name
#define SEM_TRACKLIST "/MstTRKS%s" //%s stream name
#define SEM_SESSION "/MstSess%s"
#define SEM_SESSCACHE "/MstSessCacheLock"
#define SESS_TIMEOUT 600 // Session timeout in seconds
#define SHM_CAPA "MstCapa"
#define SHM_PROTO "MstProt"
#define SHM_PROXY "MstProx"
#define SHM_STATE_LOGS "MstStateLogs"
#define SHM_STATE_ACCS "MstStateAccs"
#define SHM_STATE_STREAMS "MstStateStreams"
#define SHM_CUSTOM_VARIABLES "MstVars"
#define SHM_CAPA "/MstCapa"
#define SHM_PROTO "/MstProt"
#define SHM_PROXY "/MstProx"
#define SHM_STATE_LOGS "/MstStateLogs"
#define SHM_STATE_ACCS "/MstStateAccs"
#define SHM_STATE_STREAMS "/MstStateStreams"
#define SHM_CUSTOM_VARIABLES "/MstVars"
#define NAME_BUFFER_SIZE 200 // char buffer size for snprintf'ing shm filenames
#define SHM_SESSIONS "/MstSess"
#define SHM_SESSIONS_ITEM 165 // 4 byte crc, 100b streamname, 20b connector, 40b host, 1b sync
@ -256,7 +256,7 @@ static inline void show_stackframe(){}
#define IPC_MAX_LEN 250 // most other implementation a maximum of 251, including terminating null
#endif
#define SHM_STREAM_ENCRYPT "MstCRYP%s" //%s stream name
#define SHM_STREAM_ENCRYPT "/MstCRYP%s" //%s stream name
#define SIMUL_TRACKS 40

3
lib/dtsc.cpp
View file

@ -2308,9 +2308,6 @@ namespace DTSC{
IPC::sharedPage toErase;
toErase.init(pageName, 0, false, false);
// Set the master flag so that the page will be destroyed once it leaves scope
#if defined(__CYGWIN__) || defined(_WIN32)
IPC::releasePage(pageName);
#endif
toErase.master = true;
}
tPages.deleteRecords(1);

183
lib/shared_memory.cpp
View file

@ -1,7 +1,4 @@
#include "auth.h"
#include "bitfields.h"
#include "defines.h"
#include "procs.h"
#include "shared_memory.h"
#include "stream.h"
#include "timing.h"
@ -10,32 +7,15 @@
#include <cstdlib>
#include <cstring>
#include <fcntl.h>
#include <iostream>
#include <sys/mman.h>
#include <sys/sem.h>
#include <unistd.h>
#if defined(__CYGWIN__) || defined(_WIN32)
#include <accctrl.h>
#include <aclapi.h>
#include <windows.h>
#endif
namespace IPC{
#if defined(__CYGWIN__) || defined(_WIN32)
static std::map<std::string, sharedPage> preservedPages;
void preservePage(std::string p){preservedPages[p].init(p, 0, false, false);}
void releasePage(std::string p){preservedPages.erase(p);}
#endif
///\brief Empty semaphore constructor, clears all values
semaphore::semaphore(){
#if defined(__CYGWIN__) || defined(_WIN32)
mySem = 0;
#else
mySem = SEM_FAILED;
#endif
isLocked = 0;
}
@ -46,11 +26,7 @@ namespace IPC{
/// otherwise \param value The initial value of the semaphore if O_CREAT is given in oflag,
/// ignored otherwise
semaphore::semaphore(const char *name, int oflag, mode_t mode, unsigned int value, bool noWait){
#if defined(__CYGWIN__) || defined(_WIN32)
mySem = 0;
#else
mySem = SEM_FAILED;
#endif
isLocked = 0;
open(name, oflag, mode, value, noWait);
}
@ -60,11 +36,7 @@ namespace IPC{
///\brief Returns whether we have a valid semaphore
semaphore::operator bool() const{
#if defined(__CYGWIN__) || defined(_WIN32)
return mySem != 0;
#else
return mySem != SEM_FAILED;
#endif
}
///\brief Opens a semaphore
@ -85,33 +57,6 @@ namespace IPC{
}
int timer = 0;
while (!(*this) && timer++ < 10){
#if defined(__CYGWIN__) || defined(_WIN32)
std::string semaName = "Global\\";
semaName += (name + 1);
if (oflag & O_CREAT){
if (oflag & O_EXCL){
// attempt opening, if succes, close handle and return false;
HANDLE tmpSem = OpenMutex(SYNCHRONIZE, false, semaName.c_str());
if (tmpSem){
CloseHandle(tmpSem);
mySem = 0;
break;
}
}
SECURITY_ATTRIBUTES security = getSecurityAttributes();
mySem = CreateMutex(&security, true, semaName.c_str());
if (value){ReleaseMutex(mySem);}
}else{
mySem = OpenMutex(SYNCHRONIZE, false, semaName.c_str());
}
if (!(*this)){
if (GetLastError() == ERROR_FILE_NOT_FOUND && !noWait){// Error code 2
Util::wait(Util::expBackoffMs(timer-1, 10, 5000));
}else{
break;
}
}
#else
if (oflag & O_CREAT){
mySem = sem_open(name, oflag, mode, value);
#if defined(__APPLE__)
@ -132,21 +77,14 @@ namespace IPC{
break;
}
}
#endif
}
if (*this){myName = (char *)name;}
}
///\brief Returns the current value of the semaphore
int semaphore::getVal() const{
#if defined(__CYGWIN__) || defined(_WIN32)
LONG res;
ReleaseSemaphore(mySem, 0,
&res); // not really release.... just checking to see if I can get the value this way
#else
int res;
sem_getvalue(mySem, &res);
#endif
return res;
}
@ -159,11 +97,7 @@ namespace IPC{
#endif
return;
}
#if defined(__CYGWIN__) || defined(_WIN32)
ReleaseMutex(mySem);
#else
sem_post(mySem);
#endif
--isLocked;
#if DEBUG >= DLVL_DEVEL
if (!isLocked){
@ -186,12 +120,8 @@ namespace IPC{
#if DEBUG >= DLVL_DEVEL
uint64_t preLockTime = Util::getMicros();
#endif
#if defined(__CYGWIN__) || defined(_WIN32)
WaitForSingleObject(mySem, INFINITE);
#else
int tmp;
do{tmp = sem_wait(mySem);}while (tmp == -1 && errno == EINTR);
#endif
#if DEBUG >= DLVL_DEVEL
lockTime = Util::getMicros();
if (lockTime - preLockTime > 50000){
@ -212,15 +142,7 @@ namespace IPC{
bool semaphore::tryWait(){
if (!(*this)){return false;}
int result;
#if defined(__CYGWIN__) || defined(_WIN32)
result = WaitForSingleObject(mySem, 0); // wait at most 1ms
if (result == 0x80){
WARN_MSG("Consistency error caught on semaphore %s", myName.c_str());
result = 0;
}
#else
do{result = sem_trywait(mySem);}while (result == -1 && errno == EINTR);
#endif
isLocked += (result == 0 ? 1 : 0);
if (isLocked == 1){lockTime = Util::getMicros();}
return isLocked;
@ -231,13 +153,7 @@ namespace IPC{
bool semaphore::tryWait(uint64_t ms){
if (!(*this)){return false;}
int result;
#if defined(__CYGWIN__) || defined(_WIN32)
result = WaitForSingleObject(mySem, ms); // wait at most 1s
if (result == 0x80){
WARN_MSG("Consistency error caught on semaphore %s", myName.c_str());
result = 0;
}
#elif defined(__APPLE__)
#if defined(__APPLE__)
/// \todo (roxlu) test tryWaitOneSecond, shared_memory.cpp
uint64_t now = Util::getMicros();
uint64_t timeout = now + (ms * 1000);
@ -264,13 +180,7 @@ namespace IPC{
bool semaphore::tryWaitOneSecond(){
if (!(*this)){return false;}
int result;
#if defined(__CYGWIN__) || defined(_WIN32)
result = WaitForSingleObject(mySem, 1000); // wait at most 1s
if (result == 0x80){
WARN_MSG("Consistency error caught on semaphore %s", myName.c_str());
result = 0;
}
#elif defined(__APPLE__)
#if defined(__APPLE__)
/// \todo (roxlu) test tryWaitOneSecond, shared_memory.cpp
uint64_t now = Util::getMicros();
uint64_t timeout = now + 1e6;
@ -295,13 +205,8 @@ namespace IPC{
void semaphore::close(){
if (*this){
while (isLocked){post();}
#if defined(__CYGWIN__) || defined(_WIN32)
CloseHandle(mySem);
mySem = 0;
#else
sem_close(mySem);
mySem = SEM_FAILED;
#endif
}
myName.clear();
}
@ -310,61 +215,22 @@ namespace IPC{
/// Intended to be called from forked child processes, to drop the reference to the semaphore.
void semaphore::abandon(){
if (*this){
#if defined(__CYGWIN__) || defined(_WIN32)
CloseHandle(mySem);
mySem = 0;
#else
sem_close(mySem);
mySem = SEM_FAILED;
#endif
}
myName.clear();
}
/// Unlinks the previously opened semaphore, closing it (if open) in the process.
void semaphore::unlink(){
#if defined(__CYGWIN__) || defined(_WIN32)
while (isLocked){post();}
#endif
#if !defined(__CYGWIN__) && !defined(_WIN32)
if (myName.size()){sem_unlink(myName.c_str());}
#endif
if (*this){
#if defined(__CYGWIN__) || defined(_WIN32)
CloseHandle(mySem);
mySem = 0;
#else
sem_close(mySem);
mySem = SEM_FAILED;
#endif
}
myName.clear();
}
#if defined(__CYGWIN__) || defined(_WIN32)
SECURITY_ATTRIBUTES semaphore::getSecurityAttributes(){
///\todo We really should clean this up sometime probably
/// We currently have everything static, because the result basically depends on everything
static SECURITY_ATTRIBUTES result;
static bool resultValid = false;
static SECURITY_DESCRIPTOR securityDescriptor;
if (resultValid){return result;}
InitializeSecurityDescriptor(&securityDescriptor, SECURITY_DESCRIPTOR_REVISION);
if (!SetSecurityDescriptorDacl(&securityDescriptor, TRUE, NULL, FALSE)){
FAIL_MSG("Failed to set pSecurityDescriptor: %u", GetLastError());
return result;
}
result.nLength = sizeof(SECURITY_ATTRIBUTES);
result.lpSecurityDescriptor = &securityDescriptor;
result.bInheritHandle = FALSE;
resultValid = true;
return result;
}
#endif
/// brief Creates a shared page
///\param name_ The name of the page to be created
///\param len_ The size to make the page
@ -395,24 +261,15 @@ namespace IPC{
/// Returns true if the open file still exists.
bool sharedPage::exists(){
#if defined(__CYGWIN__) || defined(_WIN32)
return true; // Not implemented under Windows: shared memory ALWAYS exists if open!
#else
struct stat sb;
if (fstat(handle, &sb)){return false;}
return (sb.st_nlink > 0);
#endif
}
///\brief Unmaps a shared page if allowed
void sharedPage::unmap(){
if (mapped){
#if defined(__CYGWIN__) || defined(_WIN32)
// under Cygwin, the mapped location is shifted by 4 to contain the page size.
UnmapViewOfFile(mapped - 4);
#else
munmap(mapped, len);
#endif
mapped = 0;
len = 0;
}
@ -423,12 +280,8 @@ namespace IPC{
unmap();
if (handle > 0){
INSANE_MSG("Closing page %s in %s mode", name.c_str(), master ? "master" : "client");
#if defined(__CYGWIN__) || defined(_WIN32)
CloseHandle(handle);
#else
::close(handle);
if (master && name != ""){shm_unlink(name.c_str());}
#endif
handle = 0;
}
}
@ -457,37 +310,6 @@ namespace IPC{
if (name.size()){
INSANE_MSG("Opening page %s in %s mode %s auto-backoff", name.c_str(),
master ? "master" : "client", autoBackoff ? "with" : "without");
#if defined(__CYGWIN__) || defined(_WIN32)
if (master){
// Under cygwin, all pages are 4 bytes longer than claimed.
handle = CreateFileMappingA(INVALID_HANDLE_VALUE, NULL, PAGE_READWRITE, 0, len + 4, name.c_str());
}else{
int i = 0;
do{
if (i != 0){Util::wait(Util::expBackoffMs(i-1, 10, 10000));}
handle = OpenFileMappingA(FILE_MAP_ALL_ACCESS, FALSE, name.c_str());
i++;
}while (i <= 10 && !handle && autoBackoff);
}
if (!handle){
MEDIUM_MSG("%s for page %s failed with error code %u",
(master ? "CreateFileMapping" : "OpenFileMapping"), name.c_str(), GetLastError());
return;
}
mapped = (char *)MapViewOfFile(handle, FILE_MAP_ALL_ACCESS, 0, 0, 0);
if (!mapped){
FAIL_MSG("MapViewOfFile for page %s failed with error code %u", name.c_str(), GetLastError());
return;
}
// Under cygwin, the extra 4 bytes contain the real size of the page.
if (master){
Bit::htobl(mapped, len);
}else{
len = Bit::btohl(mapped);
}
// Now shift by those 4 bytes.
mapped += 4;
#else
handle = shm_open(name.c_str(), (master ? O_CREAT | O_EXCL : 0) | O_RDWR, ACCESSPERMS);
if (handle == -1){
if (master){
@ -537,7 +359,6 @@ namespace IPC{
mapped = 0;
return;
}
#endif
}
}

22
lib/shared_memory.h
View file

@ -1,16 +1,10 @@
#pragma once
#include <set>
#include <string>
#include <sys/stat.h>
#include "defines.h"
#include "timing.h"
#if defined(__CYGWIN__) || defined(_WIN32)
#include <windows.h>
#else
#include <semaphore.h>
#endif
#ifndef ACCESSPERMS
#define ACCESSPERMS (S_IRWXU | S_IRWXG | S_IRWXO)
@ -42,13 +36,7 @@ namespace IPC{
void unlink();
private:
#if defined(__CYGWIN__) || defined(_WIN32)
///\todo Maybe sometime implement anything else than 777
static SECURITY_ATTRIBUTES getSecurityAttributes();
HANDLE mySem;
#else
sem_t *mySem;
#endif
unsigned int isLocked;
uint64_t lockTime;
std::string myName;
@ -90,11 +78,6 @@ namespace IPC{
char *mapped;
};
#if defined(__CYGWIN__) || defined(_WIN32)
void preservePage(std::string);
void releasePage(std::string);
#endif
#ifdef SHM_ENABLED
///\brief A class for managing shared memory pages.
class sharedPage{
@ -109,13 +92,8 @@ namespace IPC{
void unmap();
void close();
bool exists();
#if defined(__CYGWIN__) || defined(_WIN32)
///\brief The handle of the opened shared memory page
HANDLE handle;
#else
///\brief The fd handle of the opened shared memory page
int handle;
#endif
///\brief The name of the opened shared memory page
std::string name;
///\brief The size in bytes of the opened shared memory page

7
src/io.cpp
View file

@ -162,15 +162,8 @@ namespace Mist{
snprintf(pageId, NAME_BUFFER_SIZE, SHM_TRACK_DATA, streamName.c_str(), idx, pageNumber);
std::string pageName(pageId);
IPC::sharedPage toErase;
#ifdef __CYGWIN__
toErase.init(pageName, 26 * 1024 * 1024, false, false);
#else
toErase.init(pageName, tPages.getInt("size", pageIdx), false, false);
#endif
// Set the master flag so that the page will be destroyed once it leaves scope
#if defined(__CYGWIN__) || defined(_WIN32)
IPC::releasePage(pageName);
#endif
toErase.master = true;
// Update the page on the tracks index page if needed
uint64_t firstKeyNum = tPages.getInt(firstKey, pageIdx);

1
src/utils/util_meta.cpp
View file

@ -1,6 +1,7 @@
#include <iostream>
#include <mist/shared_memory.h>
#include <mist/util.h>
#include <mist/timing.h>
int main(int argc, char **argv){
Util::redirectLogsIfNeeded();