SHS/pip
4
3
Fork 0
Code Issues 67 Pull Requests 1 Actions Releases Wiki Activity

git-svn-id: svn://db.shs.com.ru/pip@828 12ceb7fc-bf1f-11e4-8940-5bc7170c53b5

Browse Source
This commit is contained in:
Бычков Андрей
2019-06-28 11:05:43 +00:00
parent 2f96641c03
commit 175c8b0bc3
2 changed files with 33 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
src_main/io_devices/pibinarylog.cpp
View File

@@ -53,6 +53,7 @@ PIBinaryLog::PIBinaryLog() {
#endif
is_started = is_indexed = is_pause = false;
current_index = -1;
log_size = 0;
setPlaySpeed(1.);
setDefaultID(1);
setPlaySpeed(1.0);
@@ -80,6 +81,7 @@ bool PIBinaryLog::openDevice() {
is_pause = false;
index.clear();
index_pos.clear();
log_size = 0;
if (mode_ == ReadWrite) {
piCoutObj << "Error: ReadWrite mode not supported, use WriteOnly or ReadOnly";
return false;
@@ -140,6 +142,7 @@ bool PIBinaryLog::closeDevice() {
is_indexed = false;
index.clear();
index_pos.clear();
log_size = 0;
if (canWrite() && isEmpty()) {
file.remove();
return true;
@@ -251,7 +254,9 @@ PIString PIBinaryLog::createNewFile() {
void PIBinaryLog::createNewFile(const PIString &path) {
if (open(path, PIIODevice::WriteOnly)) newFile(file.path());
if (open(path, PIIODevice::WriteOnly)) {
newFile(file.path());
}
else piCoutObj << "Can't create new file, maybe path is invalid.";
}
@@ -271,9 +276,10 @@ int PIBinaryLog::writeBinLog(int id, const void *data, int size) {
piCoutObj << "Error: can`t write with id = 0!";
return -1;
}
logmutex.lock();
switch (split_mode) {
case SplitSize:
if (logSize() > split_size) createNewFile();
if (file.size() > split_size) createNewFile();
break;
case SplitTime:
if ((PISystemTime::current() - startlogtime) > split_time) createNewFile();
@@ -283,12 +289,17 @@ int PIBinaryLog::writeBinLog(int id, const void *data, int size) {
break;
default: break;
}
if (is_pause) return 0;
if (is_pause) {
logmutex.unlock();
return 0;
}
PIByteArray logdata;
logdata << id << size << (PISystemTime::current() - startlogtime) << PIByteArray::RawData(data, size);
int res = file.write(logdata.data(), logdata.size());
file.flush();
write_count++;
log_size = file.size();
logmutex.unlock();
if (res > 0) return size;
else return res;
}
@@ -298,9 +309,12 @@ int PIBinaryLog::writeBinLog_raw(int id, const PISystemTime &time, const void *d
if (size <= 0 || !canWrite()) return -1;
PIByteArray logdata;
logdata << id << size << time << PIByteArray::RawData(data, size);
logmutex.lock();
int res = file.write(logdata.data(), logdata.size());
file.flush();
write_count++;
log_size = file.size();
logmutex.unlock();
if (res > 0) return size;
else return res;
}
@@ -408,7 +422,10 @@ bool PIBinaryLog::checkFileHeader() {
}
uchar read_version = 0;
if (file.read(&read_version, 1) < 0) return false;
if (read_version == PIBINARYLOG_VERSION) return true;
if (read_version == PIBINARYLOG_VERSION) {
log_size = file.size();
return true;
}
if (read_version == 0)
piCoutObj << "BinLogFile has invalid version";
if (read_version < PIBINARYLOG_VERSION)
@@ -443,9 +460,7 @@ PIBinaryLog::BinLogRecord PIBinaryLog::readRecord() {
} else br.id = 0;
lastrecord = br;
if (br.id == 0) fileError();
file_mutex.lock();
moveIndex(index_pos.value(file.pos(), -1));
file_mutex.unlock();
logmutex.unlock();
// piCoutObj << "readRecord done";
return br;
@@ -453,7 +468,7 @@ PIBinaryLog::BinLogRecord PIBinaryLog::readRecord() {
void PIBinaryLog::parseLog(PIFile * f, PIBinaryLog::BinLogInfo * info, PIVector<PIBinaryLog::BinLogIndex> * index, PIMutex & file_mutex) {
void PIBinaryLog::parseLog(PIFile * f, PIBinaryLog::BinLogInfo * info, PIVector<PIBinaryLog::BinLogIndex> * index) {
BinLogInfo * bi = info;
bool ginfo = info != 0;
bool gindex = index != 0;
@@ -466,7 +481,6 @@ void PIBinaryLog::parseLog(PIFile * f, PIBinaryLog::BinLogInfo * info, PIVector<
if (gindex) index->clear();
if (f == 0) return;
if (!f->canRead()) return;
file_mutex.lock();
if (ginfo) {
bi->path = f->path();
bi->log_size = f->size();
@@ -479,7 +493,6 @@ void PIBinaryLog::parseLog(PIFile * f, PIBinaryLog::BinLogInfo * info, PIVector<
if (read_sig[i] != __S__PIBinaryLog::binlog_sig[i]) {if (ginfo) bi->records_count = -2; ok = false;}
uchar read_version = 0;
if (f->read(&read_version, 1) < 0) {if (ginfo) bi->records_count = -3; ok = false;}
file_mutex.unlock();
if (read_version == 0) {if (ginfo) bi->records_count = -4; ok = false;}
if (read_version < PIBINARYLOG_VERSION) {if (ginfo) bi->records_count = -5; ok = false;}
if (read_version > PIBINARYLOG_VERSION) {if (ginfo) bi->records_count = -6; ok = false;}
@@ -492,12 +505,11 @@ void PIBinaryLog::parseLog(PIFile * f, PIBinaryLog::BinLogInfo * info, PIVector<
while (1) {
ba.resize(hdr_size);
{
PIMutexLocker _ml(file_mutex);
if (f->read(ba.data(), ba.size()) > 0) {
ba >> br.id >> br.size >> br.timestamp;
} else
break;
if (f->size() - f->pos() >= br.size)
if (bi->log_size - f->pos() >= br.size)
f->seek(f->pos() + br.size);
else
break;
@@ -506,9 +518,7 @@ void PIBinaryLog::parseLog(PIFile * f, PIBinaryLog::BinLogInfo * info, PIVector<
if (gindex) {
BinLogIndex bl_ind;
bl_ind.id = br.id;
file_mutex.lock();
bl_ind.pos = f->pos() - br.size - hdr_size;
file_mutex.unlock();
bl_ind.timestamp = br.timestamp;
index->append(bl_ind);
}
@@ -550,8 +560,7 @@ PIBinaryLog::BinLogInfo PIBinaryLog::getLogInfo(const PIString & path) {
bi.records_count = 0;
PIFile tfile;
if (!tfile.open(path, PIIODevice::ReadOnly)) return bi;
PIMutex _m;
parseLog(&tfile, &bi, 0, _m);
parseLog(&tfile, &bi, 0);
return bi;
}
@@ -581,18 +590,16 @@ bool PIBinaryLog::cutBinLog(const PIBinaryLog::BinLogInfo & src, const PIString
bool PIBinaryLog::createIndex() {
file_mutex.lock();
logmutex.lock();
llong cp = file.pos();
file.seekToBegin();
file_mutex.unlock();
index.clear();
index_pos.clear();
parseLog(&file, &binfo, &index, file_mutex);
file_mutex.lock();
parseLog(&file, &binfo, &index);
file.seek(cp);
file_mutex.unlock();
is_indexed = !index.isEmpty();
for (uint i=0; i<index.size(); i++) index_pos[index[i].pos] = i;
logmutex.unlock();
return is_indexed;
}
@@ -602,9 +609,7 @@ void PIBinaryLog::seekTo(int rindex) {
logmutex.lock();
if (rindex < index.size_s() && rindex >= 0) {
file.seek(index[rindex].pos);
file_mutex.lock();
moveIndex(index_pos.value(file.pos(), -1));
file_mutex.unlock();
play_time = index[rindex].timestamp.toMilliseconds();
lastrecord.timestamp = index[rindex].timestamp;
}

12
src_main/io_devices/pibinarylog.h
View File

@@ -187,16 +187,16 @@ public:
int readBinLog(int id, void * read_to, int max_size, PISystemTime * time = 0);
//! Returns binary log file size
llong logSize() const {PIMutexLocker _ml(file_mutex); return file.size();}
llong logSize() const {return log_size;}
//! Return position in current binlog file
llong logPos() const {PIMutexLocker _ml(file_mutex); return file.pos();}
llong logPos() const {return file.pos();}
//! Return true, if position at the end of BinLog file
bool isEnd() const {if (isClosed()) return true; return file.isEnd();}
//! Returns if BinLog file is empty
bool isEmpty() const {return (file.size() <= llong(PIBINARYLOG_SIGNATURE_SIZE + 1));}
bool isEmpty() const {return (log_size <= llong(PIBINARYLOG_SIGNATURE_SIZE + 1));}
//! Returns BinLog pause status
bool isPause() const {return is_pause;}
@@ -303,7 +303,7 @@ private:
bool writeFileHeader();
bool checkFileHeader();
BinLogRecord readRecord();
static void parseLog(PIFile *f, BinLogInfo *info, PIVector<BinLogIndex> * index, PIMutex & file_mutex);
static void parseLog(PIFile *f, BinLogInfo *info, PIVector<BinLogIndex> * index);
void moveIndex(int i);
PIString getLogfilePath() const;
@@ -316,9 +316,9 @@ private:
PIFile file;
BinLogRecord lastrecord;
PISystemTime startlogtime, play_delay, split_time, pause_time;
mutable PIMutex logmutex, pausemutex, file_mutex;
mutable PIMutex logmutex, pausemutex;
double play_time, play_speed;
llong split_size;
llong split_size, log_size;
int write_count, split_count, default_id, current_index;
bool is_started, is_thread_ok, is_indexed, rapid_start, is_pause;
};