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
  31 static std::mutex launchMutex;
  32
  33 namespace grk_plugin {
  34
  35 struct GrokLogger : public MessengerLogger
  36 {
  37         explicit GrokLogger(const std::string& preamble)
  38                 : MessengerLogger(preamble)
  39         {
  40         }
  41
  42         virtual ~GrokLogger() = default;
  43
  44         void info(const char* fmt, ...) override
  45         {
  46                 va_list arg;
  47                 va_start(arg, fmt);
  48                 dcpomatic_log->log(preamble_ + log_message(fmt, arg), LogEntry::TYPE_GENERAL);
  49                 va_end(arg);
  50         }
  51
  52         void warn(const char* fmt, ...) override
  53         {
  54                 va_list arg;
  55                 va_start(arg, fmt);
  56                 dcpomatic_log->log(preamble_ + log_message(fmt, arg), LogEntry::TYPE_WARNING);
  57                 va_end(arg);
  58         }
  59
  60         void error(const char* fmt, ...) override
  61         {
  62                 va_list arg;
  63                 va_start(arg, fmt);
  64                 dcpomatic_log->log(preamble_ + log_message(fmt, arg), LogEntry::TYPE_ERROR);
  65                 va_end(arg);
  66         }
  67 };
  68
  69 struct GrokInitializer
  70 {
  71         GrokInitializer()
  72         {
  73                 setMessengerLogger(new GrokLogger("[GROK] "));
  74         }
  75
  76         ~GrokInitializer() = default;
  77 };
  78
  79 struct FrameProxy
  80 {
  81         FrameProxy()
  82                 : FrameProxy(0, Eyes::LEFT, DCPVideo())
  83         {
  84         }
  85
  86         FrameProxy(int index, Eyes eyes, DCPVideo dcpv)
  87                 : index_(index)
  88                 , eyes_(eyes)
  89                 , vf(dcpv)
  90         {
  91         }
  92
  93         int index() const
  94         {
  95                 return index_;
  96         }
  97
  98         Eyes eyes() const
  99         {
 100                 return eyes_;
 101         }
 102
 103         int index_;
 104         Eyes eyes_;
 105         DCPVideo vf;
 106 };
 107
 108 struct DcpomaticContext
 109 {
 110         DcpomaticContext(std::shared_ptr<const Film> film, Writer& writer,
 111                          EventHistory& history, const std::string& location)
 112                 : film_(film)
 113                 , writer_(writer)
 114                 , history_(history)
 115                 , location_(location)
 116                 , width_(0)
 117                 , height_(0)
 118         {
 119         }
 120
 121         void set_dimensions(uint32_t w, uint32_t h)
 122         {
 123                 width_ = w;
 124                 height_ = h;
 125         }
 126
 127         std::shared_ptr<const Film> film_;
 128         Writer& writer_;
 129         EventHistory& history_;
 130         std::string location_;
 131         uint32_t width_;
 132         uint32_t height_;
 133 };
 134
 135 class GrokContext
 136 {
 137 public:
 138         explicit GrokContext(const DcpomaticContext& dcpomaticContext)
 139                 : _dcpomatic_context(dcpomaticContext)
 140                 , _messenger(nullptr)
 141                 , _launched(false)
 142         {
 143                 struct CompressedData : public dcp::Data
 144                 {
 145                         explicit CompressedData(int dataLen)
 146                                 : data_(new uint8_t[dataLen])
 147                                 , dataLen_(dataLen)
 148                         {
 149                         }
 150
 151                         ~CompressedData()
 152                         {
 153                                 delete[] data_;
 154                         }
 155
 156                         uint8_t const* data() const override
 157                         {
 158                                 return data_;
 159                         }
 160
 161                         uint8_t* data() override
 162                         {
 163                                 return data_;
 164                         }
 165
 166                         int size() const override
 167                         {
 168                                 return dataLen_;
 169                         }
 170
 171                         uint8_t* data_;
 172                         int dataLen_;
 173                 };
 174
 175                 if (Config::instance()->enable_gpu()) {
 176                         boost::filesystem::path folder(_dcpomatic_context.location_);
 177                         boost::filesystem::path binaryPath = folder / "grk_compress";
 178                         if (!boost::filesystem::exists(binaryPath)) {
 179                                 getMessengerLogger()->error("Invalid binary location %s",
 180                                                             _dcpomatic_context.location_.c_str());
 181                                 return;
 182                         }
 183                         auto proc = [this](const std::string& str) {
 184                                 try {
 185                                         Msg msg(str);
 186                                         auto tag = msg.next();
 187                                         if (tag == GRK_MSGR_BATCH_SUBMIT_COMPRESSED) {
 188                                                 auto clientFrameId = msg.next_uint();
 189                                                 auto compressedFrameId = msg.next_uint();
 190                                                 (void)compressedFrameId;
 191                                                 auto compressedFrameLength = msg.next_uint();
 192                                                 auto processor =
 193                                                     [this](FrameProxy srcFrame, uint8_t* compressed, uint32_t compressedFrameLength) {
 194                                                             auto compressedData = std::make_shared<CompressedData>(compressedFrameLength);
 195                                                             memcpy(compressedData->data_, compressed, compressedFrameLength);
 196                                                             _dcpomatic_context.writer_.write(compressedData, srcFrame.index(), srcFrame.eyes());
 197                                                             frame_done();
 198                                                     };
 199                                                 int const minimum_size = 16384;
 200                                                 bool needsRecompression = compressedFrameLength < minimum_size;
 201                                                 _messenger->process_compressed(str, processor, needsRecompression);
 202                                                 if (needsRecompression) {
 203                                                         bool success = false;
 204                                                         auto fp = _messenger->retrieve(clientFrameId, success);
 205                                                         if (!success) {
 206                                                                 return;
 207                                                         }
 208
 209                                                         auto encoded = std::make_shared<dcp::ArrayData>(fp.vf.encode_locally());
 210                                                         _dcpomatic_context.writer_.write(encoded, fp.vf.index(), fp.vf.eyes());
 211                                                         frame_done();
 212                                                 }
 213                                         }
 214                                 } catch (std::exception& ex) {
 215                                         getMessengerLogger()->error("%s", ex.what());
 216                                 }
 217                         };
 218                         auto clientInit =
 219                             MessengerInit(clientToGrokMessageBuf, clientSentSynch, grokReceiveReadySynch,
 220                                           grokToClientMessageBuf, grokSentSynch, clientReceiveReadySynch, proc,
 221                                           std::thread::hardware_concurrency());
 222                         _messenger = new ScheduledMessenger<FrameProxy>(clientInit);
 223                 }
 224         }
 225
 226         ~GrokContext()
 227         {
 228                 shutdown();
 229         }
 230
 231         bool launch(DCPVideo dcpv, int device)
 232         {
 233                 if (!_messenger) {
 234                         return false;
 235                 }
 236                 if (_launched) {
 237                         return true;
 238                 }
 239                 std::unique_lock<std::mutex> lk_global(launchMutex);
 240                 if (!_messenger) {
 241                         return false;
 242                 }
 243                 if (_launched) {
 244                         return true;
 245                 }
 246                 if (MessengerInit::first_launch(true)) {
 247                         auto s = dcpv.get_size();
 248                         _dcpomatic_context.set_dimensions(s.width, s.height);
 249                         auto config = Config::instance();
 250                         _messenger->launch_grok(_dcpomatic_context.location_,
 251                                                 _dcpomatic_context.width_, _dcpomatic_context.width_,
 252                                                 _dcpomatic_context.height_,
 253                                                 3, 12, device,
 254                                                 _dcpomatic_context.film_->resolution() == Resolution::FOUR_K,
 255                                                 _dcpomatic_context.film_->video_frame_rate(),
 256                                                 _dcpomatic_context.film_->j2k_bandwidth(),
 257                                                 config->gpu_license_server(),
 258                                                 config->gpu_license_port(),
 259                                                 config->gpu_license());
 260                 }
 261                 _launched = _messenger->wait_for_client_init();
 262
 263                 return _launched;
 264         }
 265
 266         bool schedule_compress(const DCPVideo& vf)
 267         {
 268                 if (!_messenger) {
 269                         return false;
 270                 }
 271
 272                 auto fp = FrameProxy(vf.index(), vf.eyes(), vf);
 273                 auto cvt = [this, &fp](BufferSrc src) {
 274                         // xyz conversion
 275                         fp.vf.convert_to_xyz((uint16_t*)src.framePtr_);
 276                 };
 277                 return _messenger->schedule_compress(fp, cvt);
 278         }
 279
 280         void shutdown()
 281         {
 282                 if (!_messenger) {
 283                         return;
 284                 }
 285
 286                 std::unique_lock<std::mutex> lk_global(launchMutex);
 287                 if (!_messenger) {
 288                         return;
 289                 }
 290                 if (_launched) {
 291                         _messenger->shutdown();
 292                 }
 293                 delete _messenger;
 294                 _messenger = nullptr;
 295         }
 296
 297         void frame_done()
 298         {
 299                 _dcpomatic_context.history_.event();
 300         }
 301
 302 private:
 303         DcpomaticContext _dcpomatic_context;
 304         ScheduledMessenger<FrameProxy>* _messenger;
 305         bool _launched;
 306 };
 307
 308 } // namespace grk_plugin