src/lib/grok_context.h

   1 /*
   2     Copyright (C) 2023 Grok Image Compression Inc.
   3
   4     This file is part of DCP-o-matic.
   5
   6     DCP-o-matic is free software; you can redistribute it and/or modify
   7     it under the terms of the GNU General Public License as published by
   8     the Free Software Foundation; either version 2 of the License, or
   9     (at your option) any later version.
  10
  11     DCP-o-matic is distributed in the hope that it will be useful,
  12     but WITHOUT ANY WARRANTY; without even the implied warranty of
  13     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14     GNU General Public License for more details.
  15
  16     You should have received a copy of the GNU General Public License
  17     along with DCP-o-matic.  If not, see <http://www.gnu.org/licenses/>.
  18
  19 */
  20
  21 #pragma once
  22
  23 #include "config.h"
  24 #include "dcpomatic_log.h"
  25 #include "grok_messenger.h"
  26 #include "log.h"
  27 #include "writer.h"
  28
  29 class Film;
  30 using dcp::Data;
  31 using namespace dcpomatic;
  32
  33 static std::mutex launchMutex;
  34
  35 namespace grk_plugin {
  36
  37 struct GrokLogger : public MessengerLogger
  38 {
  39         explicit GrokLogger(const std::string& preamble)
  40                 : MessengerLogger(preamble)
  41         {
  42         }
  43
  44         virtual ~GrokLogger() = default;
  45
  46         void info(const char* fmt, ...) override
  47         {
  48                 va_list arg;
  49                 va_start(arg, fmt);
  50                 dcpomatic_log->log(preamble_ + log_message(fmt, arg), LogEntry::TYPE_GENERAL);
  51                 va_end(arg);
  52         }
  53
  54         void warn(const char* fmt, ...) override
  55         {
  56                 va_list arg;
  57                 va_start(arg, fmt);
  58                 dcpomatic_log->log(preamble_ + log_message(fmt, arg), LogEntry::TYPE_WARNING);
  59                 va_end(arg);
  60         }
  61
  62         void error(const char* fmt, ...) override
  63         {
  64                 va_list arg;
  65                 va_start(arg, fmt);
  66                 dcpomatic_log->log(preamble_ + log_message(fmt, arg), LogEntry::TYPE_ERROR);
  67                 va_end(arg);
  68         }
  69 };
  70
  71 struct GrokInitializer
  72 {
  73         GrokInitializer()
  74         {
  75                 setMessengerLogger(new GrokLogger("[GROK] "));
  76         }
  77
  78         ~GrokInitializer() = default;
  79 };
  80
  81 struct FrameProxy
  82 {
  83         FrameProxy()
  84                 : FrameProxy(0, Eyes::LEFT, DCPVideo())
  85         {
  86         }
  87
  88         FrameProxy(int index, Eyes eyes, DCPVideo dcpv)
  89                 : index_(index)
  90                 , eyes_(eyes)
  91                 , vf(dcpv)
  92         {
  93         }
  94
  95         int index() const
  96         {
  97                 return index_;
  98         }
  99
 100         Eyes eyes() const
 101         {
 102                 return eyes_;
 103         }
 104
 105         int index_;
 106         Eyes eyes_;
 107         DCPVideo vf;
 108 };
 109
 110 struct DcpomaticContext
 111 {
 112         DcpomaticContext(std::shared_ptr<const Film> film, Writer& writer,
 113                          EventHistory& history, const std::string& location)
 114                 : film_(film)
 115                 , writer_(writer)
 116                 , history_(history)
 117                 , location_(location)
 118                 , width_(0)
 119                 , height_(0)
 120         {
 121         }
 122
 123         void set_dimensions(uint32_t w, uint32_t h)
 124         {
 125                 width_ = w;
 126                 height_ = h;
 127         }
 128
 129         std::shared_ptr<const Film> film_;
 130         Writer& writer_;
 131         EventHistory& history_;
 132         std::string location_;
 133         uint32_t width_;
 134         uint32_t height_;
 135 };
 136
 137 class GrokContext
 138 {
 139 public:
 140         explicit GrokContext(const DcpomaticContext& dcpomaticContext)
 141                 : _dcpomatic_context(dcpomaticContext)
 142                 , _messenger(nullptr)
 143                 , _launched(false)
 144         {
 145                 struct CompressedData : public dcp::Data
 146                 {
 147                         explicit CompressedData(int dataLen)
 148                                 : data_(new uint8_t[dataLen])
 149                                 , dataLen_(dataLen)
 150                         {
 151                         }
 152
 153                         ~CompressedData()
 154                         {
 155                                 delete[] data_;
 156                         }
 157
 158                         uint8_t const* data() const override
 159                         {
 160                                 return data_;
 161                         }
 162
 163                         uint8_t* data() override
 164                         {
 165                                 return data_;
 166                         }
 167
 168                         int size() const override
 169                         {
 170                                 return dataLen_;
 171                         }
 172
 173                         uint8_t* data_;
 174                         int dataLen_;
 175                 };
 176
 177                 if (Config::instance()->enable_gpu()) {
 178                         boost::filesystem::path folder(_dcpomatic_context.location_);
 179                         boost::filesystem::path binaryPath = folder / "grk_compress";
 180                         if (!boost::filesystem::exists(binaryPath)) {
 181                                 getMessengerLogger()->error("Invalid binary location %s",
 182                                                             _dcpomatic_context.location_.c_str());
 183                                 return;
 184                         }
 185                         auto proc = [this](const std::string& str) {
 186                                 try {
 187                                         Msg msg(str);
 188                                         auto tag = msg.next();
 189                                         if (tag == GRK_MSGR_BATCH_SUBMIT_COMPRESSED) {
 190                                                 auto clientFrameId = msg.next_uint();
 191                                                 auto compressedFrameId = msg.next_uint();
 192                                                 (void)compressedFrameId;
 193                                                 auto compressedFrameLength = msg.next_uint();
 194                                                 auto processor =
 195                                                     [this](FrameProxy srcFrame, uint8_t* compressed, uint32_t compressedFrameLength) {
 196                                                             auto compressedData = std::make_shared<CompressedData>(compressedFrameLength);
 197                                                             memcpy(compressedData->data_, compressed, compressedFrameLength);
 198                                                             _dcpomatic_context.writer_.write(compressedData, srcFrame.index(), srcFrame.eyes());
 199                                                             frame_done();
 200                                                     };
 201                                                 int const minimum_size = 16384;
 202                                                 bool needsRecompression = compressedFrameLength < minimum_size;
 203                                                 _messenger->process_compressed(str, processor, needsRecompression);
 204                                                 if (needsRecompression) {
 205                                                         bool success = false;
 206                                                         auto fp = _messenger->retrieve(clientFrameId, success);
 207                                                         if (!success) {
 208                                                                 return;
 209                                                         }
 210
 211                                                         auto encoded = std::make_shared<dcp::ArrayData>(fp.vf.encode_locally());
 212                                                         _dcpomatic_context.writer_.write(encoded, fp.vf.index(), fp.vf.eyes());
 213                                                         frame_done();
 214                                                 }
 215                                         }
 216                                 } catch (std::exception& ex) {
 217                                         getMessengerLogger()->error("%s", ex.what());
 218                                 }
 219                         };
 220                         auto clientInit =
 221                             MessengerInit(clientToGrokMessageBuf, clientSentSynch, grokReceiveReadySynch,
 222                                           grokToClientMessageBuf, grokSentSynch, clientReceiveReadySynch, proc,
 223                                           std::thread::hardware_concurrency());
 224                         _messenger = new ScheduledMessenger<FrameProxy>(clientInit);
 225                 }
 226         }
 227
 228         ~GrokContext()
 229         {
 230                 shutdown();
 231         }
 232
 233         bool launch(DCPVideo dcpv, int device)
 234         {
 235                 if (!_messenger) {
 236                         return false;
 237                 }
 238                 if (_launched) {
 239                         return true;
 240                 }
 241                 std::unique_lock<std::mutex> lk_global(launchMutex);
 242                 if (!_messenger) {
 243                         return false;
 244                 }
 245                 if (_launched) {
 246                         return true;
 247                 }
 248                 if (MessengerInit::first_launch(true)) {
 249                         auto s = dcpv.get_size();
 250                         _dcpomatic_context.set_dimensions(s.width, s.height);
 251                         auto config = Config::instance();
 252                         _messenger->launch_grok(_dcpomatic_context.location_,
 253                                                 _dcpomatic_context.width_, _dcpomatic_context.width_,
 254                                                 _dcpomatic_context.height_,
 255                                                 3, 12, device,
 256                                                 _dcpomatic_context.film_->resolution() == Resolution::FOUR_K,
 257                                                 _dcpomatic_context.film_->video_frame_rate(),
 258                                                 _dcpomatic_context.film_->j2k_bandwidth(),
 259                                                 config->gpu_license_server(),
 260                                                 config->gpu_license_port(),
 261                                                 config->gpu_license());
 262                 }
 263                 _launched = _messenger->wait_for_client_init();
 264
 265                 return _launched;
 266         }
 267
 268         bool schedule_compress(const DCPVideo& vf)
 269         {
 270                 if (!_messenger) {
 271                         return false;
 272                 }
 273
 274                 auto fp = FrameProxy(vf.index(), vf.eyes(), vf);
 275                 auto cvt = [this, &fp](BufferSrc src) {
 276                         // xyz conversion
 277                         fp.vf.convert_to_xyz((uint16_t*)src.framePtr_);
 278                 };
 279                 return _messenger->schedule_compress(fp, cvt);
 280         }
 281
 282         void shutdown()
 283         {
 284                 if (!_messenger) {
 285                         return;
 286                 }
 287
 288                 std::unique_lock<std::mutex> lk_global(launchMutex);
 289                 if (!_messenger) {
 290                         return;
 291                 }
 292                 if (_launched) {
 293                         _messenger->shutdown();
 294                 }
 295                 delete _messenger;
 296                 _messenger = nullptr;
 297         }
 298
 299         void frame_done()
 300         {
 301                 _dcpomatic_context.history_.event();
 302         }
 303
 304 private:
 305         DcpomaticContext _dcpomatic_context;
 306         ScheduledMessenger<FrameProxy>* _messenger;
 307         bool _launched;
 308 };
 309
 310 } // namespace grk_plugin