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(
 111                 std::shared_ptr<const Film> film_,
 112                 Writer& writer_,
 113                 EventHistory& history_,
 114                 const std::string& location_
 115                 )
 116                 : film(film_)
 117                 , writer(writer_)
 118                 , history(history_)
 119                 , location(location_)
 120                 , width(0)
 121                 , height(0)
 122         {
 123         }
 124
 125         void set_dimensions(uint32_t w, uint32_t h)
 126         {
 127                 width = w;
 128                 height = h;
 129         }
 130
 131         std::shared_ptr<const Film> film;
 132         Writer& writer;
 133         EventHistory& history;
 134         std::string location;
 135         uint32_t width;
 136         uint32_t height;
 137 };
 138
 139 class GrokContext
 140 {
 141 public:
 142         explicit GrokContext(const DcpomaticContext& dcpomaticContext)
 143                 : _dcpomatic_context(dcpomaticContext)
 144                 , _messenger(nullptr)
 145                 , _launched(false)
 146         {
 147                 struct CompressedData : public dcp::Data
 148                 {
 149                         explicit CompressedData(int dataLen)
 150                                 : data_(new uint8_t[dataLen])
 151                                 , dataLen_(dataLen)
 152                         {
 153                         }
 154
 155                         ~CompressedData()
 156                         {
 157                                 delete[] data_;
 158                         }
 159
 160                         uint8_t const* data() const override
 161                         {
 162                                 return data_;
 163                         }
 164
 165                         uint8_t* data() override
 166                         {
 167                                 return data_;
 168                         }
 169
 170                         int size() const override
 171                         {
 172                                 return dataLen_;
 173                         }
 174
 175                         uint8_t* data_;
 176                         int dataLen_;
 177                 };
 178
 179                 if (Config::instance()->enable_gpu()) {
 180                         boost::filesystem::path folder(_dcpomatic_context.location);
 181                         boost::filesystem::path binaryPath = folder / "grk_compress";
 182                         if (!boost::filesystem::exists(binaryPath)) {
 183                                 getMessengerLogger()->error("Invalid binary location %s",
 184                                                             _dcpomatic_context.location.c_str());
 185                                 return;
 186                         }
 187                         auto proc = [this](const std::string& str) {
 188                                 try {
 189                                         Msg msg(str);
 190                                         auto tag = msg.next();
 191                                         if (tag == GRK_MSGR_BATCH_SUBMIT_COMPRESSED) {
 192                                                 auto clientFrameId = msg.next_uint();
 193                                                 auto compressedFrameId = msg.next_uint();
 194                                                 (void)compressedFrameId;
 195                                                 auto compressedFrameLength = msg.next_uint();
 196                                                 auto processor =
 197                                                     [this](FrameProxy srcFrame, uint8_t* compressed, uint32_t compressedFrameLength) {
 198                                                             auto compressedData = std::make_shared<CompressedData>(compressedFrameLength);
 199                                                             memcpy(compressedData->data_, compressed, compressedFrameLength);
 200                                                             _dcpomatic_context.writer.write(compressedData, srcFrame.index(), srcFrame.eyes());
 201                                                             frame_done();
 202                                                     };
 203                                                 int const minimum_size = 16384;
 204                                                 bool needsRecompression = compressedFrameLength < minimum_size;
 205                                                 _messenger->process_compressed(str, processor, needsRecompression);
 206                                                 if (needsRecompression) {
 207                                                         bool success = false;
 208                                                         auto fp = _messenger->retrieve(clientFrameId, success);
 209                                                         if (!success) {
 210                                                                 return;
 211                                                         }
 212
 213                                                         auto encoded = std::make_shared<dcp::ArrayData>(fp.vf.encode_locally());
 214                                                         _dcpomatic_context.writer.write(encoded, fp.vf.index(), fp.vf.eyes());
 215                                                         frame_done();
 216                                                 }
 217                                         }
 218                                 } catch (std::exception& ex) {
 219                                         getMessengerLogger()->error("%s", ex.what());
 220                                 }
 221                         };
 222                         auto clientInit =
 223                             MessengerInit(clientToGrokMessageBuf, clientSentSynch, grokReceiveReadySynch,
 224                                           grokToClientMessageBuf, grokSentSynch, clientReceiveReadySynch, proc,
 225                                           std::thread::hardware_concurrency());
 226                         _messenger = new ScheduledMessenger<FrameProxy>(clientInit);
 227                 }
 228         }
 229
 230         ~GrokContext()
 231         {
 232                 shutdown();
 233         }
 234
 235         bool launch(DCPVideo dcpv, int device)
 236         {
 237                 if (!_messenger) {
 238                         return false;
 239                 }
 240                 if (_launched) {
 241                         return true;
 242                 }
 243                 std::unique_lock<std::mutex> lk_global(launchMutex);
 244                 if (!_messenger) {
 245                         return false;
 246                 }
 247                 if (_launched) {
 248                         return true;
 249                 }
 250                 if (MessengerInit::first_launch(true)) {
 251                         auto s = dcpv.get_size();
 252                         _dcpomatic_context.set_dimensions(s.width, s.height);
 253                         auto config = Config::instance();
 254                         _messenger->launch_grok(_dcpomatic_context.location,
 255                                                 _dcpomatic_context.width, _dcpomatic_context.width,
 256                                                 _dcpomatic_context.height,
 257                                                 3, 12, device,
 258                                                 _dcpomatic_context.film->resolution() == Resolution::FOUR_K,
 259                                                 _dcpomatic_context.film->video_frame_rate(),
 260                                                 _dcpomatic_context.film->j2k_bandwidth(),
 261                                                 config->gpu_license_server(),
 262                                                 config->gpu_license_port(),
 263                                                 config->gpu_license());
 264                 }
 265                 _launched = _messenger->wait_for_client_init();
 266
 267                 return _launched;
 268         }
 269
 270         bool schedule_compress(const DCPVideo& vf)
 271         {
 272                 if (!_messenger) {
 273                         return false;
 274                 }
 275
 276                 auto fp = FrameProxy(vf.index(), vf.eyes(), vf);
 277                 auto cvt = [this, &fp](BufferSrc src) {
 278                         // xyz conversion
 279                         fp.vf.convert_to_xyz((uint16_t*)src.framePtr_);
 280                 };
 281                 return _messenger->schedule_compress(fp, cvt);
 282         }
 283
 284         void shutdown()
 285         {
 286                 if (!_messenger) {
 287                         return;
 288                 }
 289
 290                 std::unique_lock<std::mutex> lk_global(launchMutex);
 291                 if (!_messenger) {
 292                         return;
 293                 }
 294                 if (_launched) {
 295                         _messenger->shutdown();
 296                 }
 297                 delete _messenger;
 298                 _messenger = nullptr;
 299         }
 300
 301         void frame_done()
 302         {
 303                 _dcpomatic_context.history.event();
 304         }
 305
 306 private:
 307         DcpomaticContext _dcpomatic_context;
 308         ScheduledMessenger<FrameProxy>* _messenger;
 309         bool _launched;
 310 };
 311
 312 } // namespace grk_plugin