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(void)
  74         {
  75                 setMessengerLogger(new GrokLogger("[GROK] "));
  76         }
  77
  78         ~GrokInitializer() = default;
  79 };
  80
  81 struct FrameProxy
  82 {
  83         FrameProxy(void)
  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(void) 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 setDimensions(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                 : dcpomaticContext_(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(void)
 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(dcpomaticContext_.location_);
 179                         boost::filesystem::path binaryPath = folder / "grk_compress";
 180                         if (!boost::filesystem::exists(binaryPath)) {
 181                                 getMessengerLogger()->error("Invalid binary location %s",
 182                                                             dcpomaticContext_.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.nextUint();
 191                                                 auto compressedFrameId = msg.nextUint();
 192                                                 (void)compressedFrameId;
 193                                                 auto compressedFrameLength = msg.nextUint();
 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                                                             dcpomaticContext_.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_->processCompressed(str, processor, needsRecompression);
 204                                                 if (needsRecompression) {
 205                                                         bool success = false;
 206                                                         auto fp = messenger_->retrieve(clientFrameId, success);
 207                                                         if (!success)
 208                                                                 return;
 209
 210                                                         auto encoded = std::make_shared<dcp::ArrayData>(fp.vf.encode_locally());
 211                                                         dcpomaticContext_.writer_.write(encoded, fp.vf.index(), fp.vf.eyes());
 212                                                         frame_done();
 213                                                 }
 214                                         }
 215                                 } catch (std::exception& ex) {
 216                                         getMessengerLogger()->error("%s", ex.what());
 217                                 }
 218                         };
 219                         auto clientInit =
 220                             MessengerInit(clientToGrokMessageBuf, clientSentSynch, grokReceiveReadySynch,
 221                                           grokToClientMessageBuf, grokSentSynch, clientReceiveReadySynch, proc,
 222                                           std::thread::hardware_concurrency());
 223                         messenger_ = new ScheduledMessenger<FrameProxy>(clientInit);
 224                 }
 225         }
 226
 227         ~GrokContext(void)
 228         {
 229                 shutdown();
 230         }
 231
 232         bool launch(DCPVideo dcpv, int device)
 233         {
 234                 if (!messenger_)
 235                         return false;
 236                 if (launched_)
 237                         return true;
 238                 std::unique_lock<std::mutex> lk_global(launchMutex);
 239                 if (!messenger_)
 240                         return false;
 241                 if (launched_)
 242                         return true;
 243                 if (MessengerInit::firstLaunch(true)) {
 244                         auto s = dcpv.get_size();
 245                         dcpomaticContext_.setDimensions(s.width, s.height);
 246                         auto config = Config::instance();
 247                         messenger_->launchGrok(dcpomaticContext_.location_,
 248                                                dcpomaticContext_.width_, dcpomaticContext_.width_,
 249                                                dcpomaticContext_.height_,
 250                                                3, 12, device,
 251                                                dcpomaticContext_.film_->resolution() == Resolution::FOUR_K,
 252                                                dcpomaticContext_.film_->video_frame_rate(),
 253                                                dcpomaticContext_.film_->j2k_bandwidth(),
 254                                                config->gpu_license_server(),
 255                                                config->gpu_license_port(),
 256                                                config->gpu_license());
 257                 }
 258                 launched_ = messenger_->waitForClientInit();
 259
 260                 return launched_;
 261         }
 262
 263         bool scheduleCompress(const DCPVideo& vf)
 264         {
 265                 if (!messenger_)
 266                         return false;
 267
 268                 auto fp = FrameProxy(vf.index(), vf.eyes(), vf);
 269                 auto cvt = [this, &fp](BufferSrc src) {
 270                         // xyz conversion
 271                         fp.vf.convert_to_xyz((uint16_t*)src.framePtr_);
 272                 };
 273                 return messenger_->scheduleCompress(fp, cvt);
 274         }
 275
 276         void shutdown(void)
 277         {
 278                 if (!messenger_)
 279                         return;
 280
 281                 std::unique_lock<std::mutex> lk_global(launchMutex);
 282                 if (!messenger_)
 283                         return;
 284                 if (launched_)
 285                         messenger_->shutdown();
 286                 delete messenger_;
 287                 messenger_ = nullptr;
 288         }
 289
 290         void frame_done()
 291         {
 292                 dcpomaticContext_.history_.event();
 293         }
 294
 295 private:
 296         DcpomaticContext dcpomaticContext_;
 297         ScheduledMessenger<FrameProxy>* messenger_;
 298         bool launched_;
 299 };
 300
 301 } // namespace grk_plugin