summaryrefslogtreecommitdiff
path: root/graphics/asymptote/LspCpp/src/jsonrpc/RemoteEndPoint.cpp
blob: b501d35a18a46bdef6ce674851e4b4f5447f03e6 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
#include "LibLsp/JsonRpc/MessageJsonHandler.h"
#include "LibLsp/JsonRpc/Endpoint.h"
#include "LibLsp/JsonRpc/message.h"
#include "LibLsp/JsonRpc/RemoteEndPoint.h"
#include <future>
#include "LibLsp/JsonRpc/Cancellation.h"
#include "LibLsp/JsonRpc/StreamMessageProducer.h"
#include "LibLsp/JsonRpc/NotificationInMessage.h"
#include "LibLsp/JsonRpc/lsResponseMessage.h"
#include "LibLsp/JsonRpc/Condition.h"
#include "LibLsp/JsonRpc/Context.h"
#include "rapidjson/error/en.h"
#include "LibLsp/JsonRpc/json.h"
#include "LibLsp/JsonRpc/GCThreadContext.h"
#include "LibLsp/JsonRpc/ScopeExit.h"
#include "LibLsp/JsonRpc/stream.h"
#include <atomic>
#include <optional>
#include <boost/asio/thread_pool.hpp>
#include <boost/asio/post.hpp>

namespace lsp {

// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
// See https://llvm.org/LICENSE.txt for license information.
// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
//===----------------------------------------------------------------------===//

// Cancellation mechanism for long-running tasks.
//
// This manages interactions between:
//
// 1. Client code that starts some long-running work, and maybe cancels later.
//
//   std::pair<Context, Canceler> Task = cancelableTask();
//   {
//     WithContext Cancelable(std::move(Task.first));
//     Expected
//     deepThoughtAsync([](int answer){ errs() << answer; });
//   }
//   // ...some time later...
//   if (User.fellAsleep())
//     Task.second();
//
//  (This example has an asynchronous computation, but synchronous examples
//  work similarly - the Canceler should be invoked from another thread).
//
// 2. Library code that executes long-running work, and can exit early if the
//   result is not needed.
//
//   void deepThoughtAsync(std::function<void(int)> Callback) {
//     runAsync([Callback]{
//       int A = ponder(6);
//       if (getCancelledMonitor())
//         return;
//       int B = ponder(9);
//       if (getCancelledMonitor())
//         return;
//       Callback(A * B);
//     });
//   }
//
//   (A real example may invoke the callback with an error on cancellation,
//   the CancelledError is provided for this purpose).
//
// Cancellation has some caveats:
//   - the work will only stop when/if the library code next checks for it.
//     Code outside clangd such as Sema will not do this.
//   - it's inherently racy: client code must be prepared to accept results
//     even after requesting cancellation.
//   - it's Context-based, so async work must be dispatched to threads in
//     ways that preserve the context. (Like runAsync() or TUScheduler).
//

        /// A canceller requests cancellation of a task, when called.
        /// Calling it again has no effect.
        using Canceler = std::function<void()>;

        // We don't want a cancelable scope to "shadow" an enclosing one.
        struct CancelState {
                std::shared_ptr<std::atomic<int>> cancelled;
                const CancelState* parent = nullptr;
                lsRequestId id;
        };
        static Key<CancelState> g_stateKey;

        /// Defines a new task whose cancellation may be requested.
        /// The returned Context defines the scope of the task.
        /// When the context is active, getCancelledMonitor() is 0 until the Canceler is
        /// invoked, and equal to Reason afterwards.
        /// Conventionally, Reason may be the LSP error code to return.
        std::pair<Context, Canceler> cancelableTask(const lsRequestId& id,int reason = 1){
                assert(reason != 0 && "Can't detect cancellation if Reason is zero");
                CancelState state;
                state.id = id;
                state.cancelled = std::make_shared<std::atomic<int>>();
                state.parent = Context::current().get(g_stateKey);
                return {
                        Context::current().derive(g_stateKey, state),
                        [reason, cancelled(state.cancelled)] { *cancelled = reason; },
                };
        }
        /// If the current context is within a cancelled task, returns the reason.
/// (If the context is within multiple nested tasks, true if any are cancelled).
/// Always zero if there is no active cancelable task.
/// This isn't free (context lookup) - don't call it in a tight loop.
        optional<CancelMonitor> getCancelledMonitor(const lsRequestId& id, const Context& ctx = Context::current()){
                for (const CancelState* state = ctx.get(g_stateKey); state != nullptr;
                        state = state->parent)
                {
                        if (id != state->id)continue;
                        const std::shared_ptr<std::atomic<int> > cancelled = state->cancelled;
                        std::function<int()> temp = [=]{
                                return cancelled->load();
                        };
                        return std::move(temp);
                }

                return {};
        }
} // namespace lsp

using namespace  lsp;
class PendingRequestInfo
{
        using   RequestCallBack = std::function< bool(std::unique_ptr<LspMessage>) >;
public:
        PendingRequestInfo(const std::string& md,
                const RequestCallBack& callback);
        PendingRequestInfo(const std::string& md);
        PendingRequestInfo() {}
        std::string method;
        RequestCallBack futureInfo;
};

PendingRequestInfo::PendingRequestInfo(const std::string& _md,
        const   RequestCallBack& callback) : method(_md),
        futureInfo(callback)
{
}

PendingRequestInfo::PendingRequestInfo(const std::string& md) : method(md)
{
}
struct RemoteEndPoint::Data
{
        explicit Data(uint8_t workers,lsp::Log& _log , RemoteEndPoint* owner)
          : max_workers(workers), m_id(0),next_request_cookie(0), message_producer(new StreamMessageProducer(*owner)), log(_log)
        {

        }
        ~Data()
        {
           delete       message_producer;
        }
    uint8_t max_workers;
        std::atomic<int> m_id;
    std::shared_ptr<boost::asio::thread_pool> tp;
        // Method calls may be cancelled by ID, so keep track of their state.
 // This needs a mutex: handlers may finish on a different thread, and that's
 // when we clean up entries in the map.
        mutable std::mutex request_cancelers_mutex;

        std::map< lsRequestId, std::pair<Canceler, /*Cookie*/ unsigned> > requestCancelers;

        std::atomic<unsigned>  next_request_cookie; // To disambiguate reused IDs, see below.
        void onCancel(Notify_Cancellation::notify* notify) {
                std::lock_guard<std::mutex> Lock(request_cancelers_mutex);
                const auto it = requestCancelers.find(notify->params.id);
                if (it != requestCancelers.end())
                        it->second.first(); // Invoke the canceler.
        }

        // We run cancelable requests in a context that does two things:
        //  - allows cancellation using requestCancelers[ID]
        //  - cleans up the entry in requestCancelers when it's no longer needed
        // If a client reuses an ID, the last wins and the first cannot be canceled.
        Context cancelableRequestContext(lsRequestId id) {
                auto task = cancelableTask(id,
                        /*Reason=*/static_cast<int>(lsErrorCodes::RequestCancelled));
                unsigned cookie;
                {
                        std::lock_guard<std::mutex> Lock(request_cancelers_mutex);
                        cookie = next_request_cookie.fetch_add(1, std::memory_order_relaxed);
                        requestCancelers[id] = { std::move(task.second), cookie };
                }
                // When the request ends, we can clean up the entry we just added.
                // The cookie lets us check that it hasn't been overwritten due to ID
                // reuse.
                return task.first.derive(lsp::make_scope_exit([this, id, cookie] {
                        std::lock_guard<std::mutex> lock(request_cancelers_mutex);
                        const auto& it = requestCancelers.find(id);
                        if (it != requestCancelers.end() && it->second.second == cookie)
                                requestCancelers.erase(it);
                        }));
        }

        std::map <lsRequestId, std::shared_ptr<PendingRequestInfo>>  _client_request_futures;
        StreamMessageProducer* message_producer;
        std::atomic<bool> quit{};
        lsp::Log& log;
        std::shared_ptr<lsp::istream>  input;
        std::shared_ptr<lsp::ostream>  output;

    std::mutex m_requestInfo;

        bool pendingRequest(RequestInMessage& info, GenericResponseHandler&& handler)
        {
        bool ret = true;
        std::lock_guard<std::mutex> lock(m_requestInfo);
        if(!info.id.has_value()){
            auto id = getNextRequestId();
            info.id.set(id);
        }
        else{
            if(_client_request_futures.find(info.id) != _client_request_futures.end()){
                ret =  false;
            }
        }
        _client_request_futures[info.id] = std::make_shared<PendingRequestInfo>(info.method, handler);
        return ret;
        }
        const std::shared_ptr<const PendingRequestInfo> getRequestInfo(const lsRequestId& _id)
        {
                std::lock_guard<std::mutex> lock(m_requestInfo);
                auto findIt = _client_request_futures.find(_id);
                if (findIt != _client_request_futures.end())
                {
                        return findIt->second;
                }
                return  nullptr;
        }

        void removeRequestInfo(const lsRequestId& _id)
        {
                std::lock_guard<std::mutex> lock(m_requestInfo);
                auto findIt = _client_request_futures.find(_id);
                if (findIt != _client_request_futures.end())
                {
                        _client_request_futures.erase(findIt);
                }
        }
        void clear()
        {
                {
                        std::lock_guard<std::mutex> lock(m_requestInfo);
                        _client_request_futures.clear();
                }
        if(tp){
            tp->stop();
        }
                quit.store(true, std::memory_order_relaxed);
        }

    int getNextRequestId()
    {
        return m_id.fetch_add(1, std::memory_order_relaxed);
    }
};

namespace
{
void WriterMsg(std::shared_ptr<lsp::ostream>&  output, LspMessage& msg)
{
        const auto& s = msg.ToJson();
        const auto value =
                std::string("Content-Length: ") + std::to_string(s.size()) + "\r\n\r\n" + s;
        output->write(value);
        output->flush();
}

bool isResponseMessage(JsonReader& visitor)
{

        if (!visitor.HasMember("id"))
        {
                return false;
        }

        if (!visitor.HasMember("result") && !visitor.HasMember("error"))
        {
                return false;
        }

        return true;
}

bool isRequestMessage(JsonReader& visitor)
{
        if (!visitor.HasMember("method"))
        {
                return false;
        }
        if (!visitor["method"]->IsString())
        {
                return false;
        }
        if (!visitor.HasMember("id"))
        {
                return false;
        }
        return true;
}
bool isNotificationMessage(JsonReader& visitor)
{
        if (!visitor.HasMember("method"))
        {
                return false;
        }
        if (!visitor["method"]->IsString())
        {
                return false;
        }
        if (visitor.HasMember("id"))
        {
                return false;
        }
        return true;
}
}

CancelMonitor RemoteEndPoint::getCancelMonitor(const lsRequestId& id)
{
        auto  monitor =  getCancelledMonitor(id);
        if(monitor.has_value())
        {
                return  monitor.value();
        }
        return [] {
                return 0;
        };

}

RemoteEndPoint::RemoteEndPoint(
        const std::shared_ptr < MessageJsonHandler >& json_handler,const std::shared_ptr < Endpoint>& localEndPoint, lsp::Log& _log, uint8_t max_workers):
    d_ptr(new Data(max_workers,_log,this)),jsonHandler(json_handler), local_endpoint(localEndPoint)
{
        jsonHandler->method2notification[Notify_Cancellation::notify::kMethodInfo] = [](Reader& visitor)
        {
                return Notify_Cancellation::notify::ReflectReader(visitor);
        };

        d_ptr->quit.store(false, std::memory_order_relaxed);

}

RemoteEndPoint::~RemoteEndPoint()
{
        delete d_ptr;
        d_ptr->quit.store(true, std::memory_order_relaxed);
}

bool RemoteEndPoint::dispatch(const std::string& content)
{
                rapidjson::Document document;
                document.Parse(content.c_str(), content.length());
                if (document.HasParseError())
                {
                        std::string info ="lsp msg format error:";
                        rapidjson::GetParseErrorFunc GetParseError = rapidjson::GetParseError_En; // or whatever
                        info+= GetParseError(document.GetParseError());
                        info += "\n";
                        info += "ErrorContext offset:\n";
                        info += content.substr(document.GetErrorOffset());
                        d_ptr->log.log(Log::Level::SEVERE, info);

                        return false;
                }

                JsonReader visitor{ &document };
                if (!visitor.HasMember("jsonrpc") ||
                        std::string(visitor["jsonrpc"]->GetString()) != "2.0")
                {
                        std::string reason;
                        reason = "Reason:Bad or missing jsonrpc version\n";
                        reason += "content:\n" + content;
                        d_ptr->log.log(Log::Level::SEVERE, reason);
                        return  false;

                }
                LspMessage::Kind _kind = LspMessage::NOTIFICATION_MESSAGE;
                try {
                        if (isRequestMessage(visitor))
                        {
                                _kind = LspMessage::REQUEST_MESSAGE;
                                auto msg = jsonHandler->parseRequstMessage(visitor["method"]->GetString(), visitor);
                                if (msg) {
                                        mainLoop(std::move(msg));
                                }
                                else {
                                        std::string info = "Unknown support request message when consumer message:\n";
                                        info += content;
                                        d_ptr->log.log(Log::Level::WARNING, info);
                                        return false;
                                }
                        }
                        else if (isResponseMessage(visitor))
                        {
                                _kind = LspMessage::RESPONCE_MESSAGE;
                                lsRequestId id;
                                ReflectMember(visitor, "id", id);

                                auto msgInfo = d_ptr->getRequestInfo(id);
                                if (!msgInfo)
                                {
                    std::string info = "Unknown response message :\n";
                    info += content;
                    d_ptr->log.log(Log::Level::INFO, info);
                                }
                                else
                                {

                                        auto msg = jsonHandler->parseResponseMessage(msgInfo->method, visitor);
                                        if (msg) {
                                                mainLoop(std::move(msg));
                                        }
                                        else
                                        {
                                                std::string info = "Unknown response message :\n";
                                                info += content;
                                                d_ptr->log.log(Log::Level::SEVERE, info);
                                                return  false;
                                        }

                                }
                        }
                        else if (isNotificationMessage(visitor))
                        {
                                auto msg = jsonHandler->parseNotificationMessage(visitor["method"]->GetString(), visitor);
                                if (!msg)
                                {
                                        std::string info = "Unknown notification message :\n";
                                        info += content;
                                        d_ptr->log.log(Log::Level::SEVERE, info);
                                        return  false;
                                }
                                mainLoop(std::move(msg));
                        }
                        else
                        {
                                std::string info = "Unknown lsp message when consumer message:\n";
                                info += content;
                                d_ptr->log.log(Log::Level::WARNING, info);
                                return false;
                        }
                }
                catch (std::exception& e)
                {

                        std::string info = "Exception  when process ";
                        if(_kind==LspMessage::REQUEST_MESSAGE)
                        {
                                info += "request";
                        }
                        if (_kind == LspMessage::RESPONCE_MESSAGE)
                        {
                                info += "response";
                        }
                        else
                        {
                                info += "notification";
                        }
                        info += " message:\n";
                        info += e.what();
                        std::string reason = "Reason:" + info + "\n";
                        reason += "content:\n" + content;
                        d_ptr->log.log(Log::Level::SEVERE, reason);
                        return false;
                }
        return  true;
}



bool RemoteEndPoint::internalSendRequest(RequestInMessage& info, GenericResponseHandler handler)
{
        std::lock_guard<std::mutex> lock(m_sendMutex);
        if (!d_ptr->output || d_ptr->output->bad())
        {
                std::string desc = "Output isn't good any more:\n";
                d_ptr->log.log(Log::Level::WARNING, desc);
                return false;
        }
        if(!d_ptr->pendingRequest(info, std::move(handler)))
    {
        std::string desc = "Duplicate id  which of request:";
        desc += info.ToJson();
        desc += "\n";
        d_ptr->log.log(Log::Level::WARNING, desc);
    }
        WriterMsg(d_ptr->output, info);
    return true;
}

int RemoteEndPoint::getNextRequestId(){
    return   d_ptr->getNextRequestId();
}
bool RemoteEndPoint::cancelRequest(const lsRequestId& id){
    if(!isWorking()){
        return false;
    }
    auto msgInfo = d_ptr->getRequestInfo(id);
    if (msgInfo){
        Notify_Cancellation::notify cancel_notify;
        cancel_notify.params.id = id;
        send(cancel_notify);
        return true;
    }
    return false;
}
std::unique_ptr<LspMessage> RemoteEndPoint::internalWaitResponse(RequestInMessage& request, unsigned time_out)
{
        auto  eventFuture = std::make_shared< Condition< LspMessage > >();
        internalSendRequest(request, [=](std::unique_ptr<LspMessage> data)
        {
                eventFuture->notify(std::move(data));
                return  true;
        });
        return   eventFuture->wait(time_out);
}

void RemoteEndPoint::mainLoop(std::unique_ptr<LspMessage>msg)
{
        if(d_ptr->quit.load(std::memory_order_relaxed))
        {
                return;
        }
        const auto _kind = msg->GetKid();
        if (_kind == LspMessage::REQUEST_MESSAGE)
        {
                auto req = static_cast<RequestInMessage*>(msg.get());
                // Calls can be canceled by the client. Add cancellation context.
                WithContext WithCancel(d_ptr->cancelableRequestContext(req->id));
                local_endpoint->onRequest(std::move(msg));
        }

        else if (_kind == LspMessage::RESPONCE_MESSAGE)
        {
                const auto id = static_cast<ResponseInMessage*>(msg.get())->id;
                auto msgInfo = d_ptr->getRequestInfo(id);
                if (!msgInfo)
                {
                        const auto _method_desc = msg->GetMethodType();
                        local_endpoint->onResponse(_method_desc, std::move(msg));
                }
                else
                {
                        bool needLocal = true;
                        if (msgInfo->futureInfo)
                        {
                                if (msgInfo->futureInfo(std::move(msg)))
                                {
                                        needLocal = false;
                                }
                        }
                        if (needLocal)
                        {
                                local_endpoint->onResponse(msgInfo->method, std::move(msg));
                        }
                        d_ptr->removeRequestInfo(id);
                }
        }
        else if (_kind == LspMessage::NOTIFICATION_MESSAGE)
        {
                if (strcmp(Notify_Cancellation::notify::kMethodInfo, msg->GetMethodType())==0)
                {
                        d_ptr->onCancel(static_cast<Notify_Cancellation::notify*>(msg.get()));
                }
                else
                {
                        local_endpoint->notify(std::move(msg));
                }

        }
        else
        {
                std::string info = "Unknown lsp message  when process  message  in mainLoop:\n";
                d_ptr->log.log(Log::Level::WARNING, info);
        }
}

void RemoteEndPoint::handle(std::vector<MessageIssue>&& issue)
{
        for(auto& it : issue)
        {
                d_ptr->log.log(it.code, it.text);
        }
}

void RemoteEndPoint::handle(MessageIssue&& issue)
{
        d_ptr->log.log(issue.code, issue.text);
}


void RemoteEndPoint::startProcessingMessages(std::shared_ptr<lsp::istream> r,
        std::shared_ptr<lsp::ostream> w)
{
        d_ptr->quit.store(false, std::memory_order_relaxed);
        d_ptr->input = r;
        d_ptr->output = w;
        d_ptr->message_producer->bind(r);
    d_ptr->tp = std::make_shared<boost::asio::thread_pool>(d_ptr->max_workers);
        message_producer_thread_ = std::make_shared<std::thread>([&]()
   {
                d_ptr->message_producer->listen([&](std::string&& content){
                        const auto temp = std::make_shared<std::string>(std::move(content));
            boost::asio::post(*d_ptr->tp,
                        [this, temp]{
#ifdef USEGC
                        GCThreadContext gcContext;
#endif

                                                dispatch(*temp);
                                });
                });
        });
}

void RemoteEndPoint::stop()
{
        if(message_producer_thread_ && message_producer_thread_->joinable())
        {
                message_producer_thread_->detach();
        message_producer_thread_ = nullptr;
        }
        d_ptr->clear();

}

void RemoteEndPoint::sendMsg( LspMessage& msg)
{

        std::lock_guard<std::mutex> lock(m_sendMutex);
        if (!d_ptr->output || d_ptr->output->bad())
        {
                std::string info = "Output isn't good any more:\n";
                d_ptr->log.log(Log::Level::INFO, info);
                return;
        }
        WriterMsg(d_ptr->output, msg);

}

bool RemoteEndPoint::isWorking() const {
    if (message_producer_thread_ && message_producer_thread_->joinable())
        return true;
    return  false;
}