Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Working #20

Merged
merged 3 commits into from
Nov 17, 2020
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 2 additions & 4 deletions epf/Cell.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,7 @@ void Cell::initialize()
m_buf = m_writer->bufferCache().fetch();
m_pos = m_buf->data();

// The end position is one less than the buffer size to allow for
// speculative writes into the cell. See FileProcessor for details.
m_endPos = m_pos + m_pointSize * ((BufSize / m_pointSize) - 1);
m_endPos = m_pos + m_pointSize * (BufSize / m_pointSize);
}

void Cell::write()
Expand All @@ -43,7 +41,7 @@ void Cell::write()
void Cell::advance()
{
m_pos += m_pointSize;
if (m_pos > m_endPos)
if (m_pos >= m_endPos)
{
write();
initialize();
Expand Down
13 changes: 6 additions & 7 deletions epf/Epf.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -158,20 +158,19 @@ void Epf::run(const std::vector<std::string>& options)
// Make a writer with 4 threads.
m_writer.reset(new Writer(m_outputDir, 4));

std::vector<std::unique_ptr<FileProcessor>> processors;

// Sort file infos so the largest files come first. This helps to make sure we don't delay
// processing big files that take the longest (use threads more efficiently).
std::sort(fileInfos.begin(), fileInfos.end(), [](const FileInfo& f1, const FileInfo& f2)
{ return f1.numPoints > f2.numPoints; });
// Add the files to the processing pool
for (const FileInfo& fi : fileInfos)
{
std::unique_ptr<FileProcessor> fp(
new FileProcessor(fi, layout->pointSize(), m_grid, m_writer.get()));
std::function<void()> processor = std::bind(&FileProcessor::operator(), fp.get());
m_pool.add(processor);
processors.push_back(std::move(fp));
int pointSize = layout->pointSize();
m_pool.add([&fi, pointSize, this]()
{
FileProcessor fp(fi, pointSize, m_grid, m_writer.get());
fp.run();
});
}

// Wait for all the processors to finish and restart.
Expand Down
2 changes: 1 addition & 1 deletion epf/FileProcessor.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ FileProcessor::FileProcessor(const FileInfo& fi, size_t pointSize, const Grid& g
m_fi(fi), m_cellMgr(pointSize, writer), m_grid(grid), m_cnt(++m_totalCnt)
{}

void FileProcessor::operator()()
void FileProcessor::run()
{
Options opts;
opts.add("filename", m_fi.filename);
Expand Down
2 changes: 1 addition & 1 deletion epf/FileProcessor.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ class FileProcessor
FileProcessor(const FileInfo& fi, size_t pointSize, const Grid& grid, Writer *writer);

Cell *getCell(const VoxelKey& key);
void operator()();
void run();

private:
FileInfo m_fi;
Expand Down
3 changes: 3 additions & 0 deletions epf/Writer.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,9 @@ void Writer::run()
// Remove the key from the active key list.
std::ofstream out(path(wd.key), std::ios::app | std::ios::binary);
out.write(reinterpret_cast<const char *>(wd.data->data()), wd.data->size());
out.close();
if (!out)
throw Error("Failure writing to '" + path(wd.key) + "'.");
m_bufferCache.replace(std::move(wd.data));

std::lock_guard<std::mutex> lock(m_mutex);
Expand Down