Revert "Use 'lparallels'. Experimental"

This reverts commit 87bf719032663e9f023f8cb0310b60615caec5ef.

chatikbot.asd

@@ -11,7 +11,6 @@
                #:drakma
                #:flexi-streams
                #:local-time
-               #:lparallel
                #:log4cl
                #:plump
                #:sqlite

chatikbot.lisp

@@ -20,7 +20,7 @@
      do (setf *telegram-last-update*
               (max (or *telegram-last-update* 0)
                    (aget "update_id" update)))
-     do (lparallel:future (handle-message (aget "message" update)))))
+     do (handle-message (aget "message" update))))
 
 
 (defun send-response (chat-id response &optional reply-id)
@@ -152,11 +152,10 @@
 (defun send-akb (text)
   (log:info "send-akb: ~A" text)
   (dolist (chat-id *akb-send-to*)
-    (lparallel:future
-      (handler-case
-          (telegram-send-message chat-id text
-                                 :disable-web-preview 1)
-        (error (e) (log:error e))))))
+    (handler-case
+        (telegram-send-message chat-id text
+                               :disable-web-preview 1)
+      (error (e) (log:error e)))))
 
 (defun handle-cmd-akb (chat-id message-id args)
   (log:info "handle-cmd-akb" chat-id message-id args)
@@ -179,16 +178,11 @@
 ;; Finance
 (defun process-rates ()
   (handler-case
-      (lparallel:plet ((ts (local-time:timestamp-to-unix (local-time:now)))
-                       (rates (get-rates))
-                       (brent (get-brent))
-                       (btc (get-btc-e)))
-        (db-add-finance ts
-                        (aget "USD/RUB" rates)
-                        (aget "EUR/RUB" rates)
-                        (aget "GBP/RUB" rates)
-                        brent
-                        btc))
+      (let ((ts (local-time:timestamp-to-unix (local-time:now)))
+            (rates (get-rates))
+            (brent (get-brent))
+            (btc (get-btc-e)))
+        (db-add-finance ts (aget "USD/RUB" rates) (aget "EUR/RUB" rates) (aget "GBP/RUB" rates) brent btc))
     (error (e) (log:error e))))
 
 (defun handle-cmd-rates (chat-id message-id args)
@@ -337,9 +331,8 @@
                       (gethash chat-id checkins)))
               (db-fsq-add-seen id created-at))))
         (loop for chat-id being the hash-keys in checkins using (hash-value texts)
-           do (lparallel:future
-                (log:info "Sending checkins" chat-id texts)
-                (telegram-send-message chat-id (format nil "~{~A~^~%~}" texts)))))
+           do (log:info "Sending checkins" chat-id texts)
+             (telegram-send-message chat-id (format nil "~{~A~^~%~}" texts))))
     (error (e) (log:error e))))
 
 
@@ -428,24 +421,16 @@
       (log:error e)
       (telegram-send-message chat-id "Ошибочка вышла"))))
 
-(defun process-feed (feed)
+(defun process-feeds ()
   (handler-case
-      (progn
+      (dolist (feed (remove-if-not #'need-fetch-p (db-rss-get-active-feeds)))
         (log:info "Fetching new items" (feed-url feed))
         (dolist (item (%fetch-new-items feed))
-          (lparallel:pmapc #'(lambda (chat-id)
-                               (telegram-send-message chat-id
-                                                      (format-feed-item item)
-                                                      :disable-web-preview 1))
-                           (db-rss-get-feed-chats feed)))
-        ;; Update next fetch and period
-        (db-rss-update-feed feed))
-    (error (e) (log:error e))))
-
-(defun process-feeds ()
-  (handler-case
-      (lparallel:pmapc 'process-feed
-                       (remove-if-not #'need-fetch-p (db-rss-get-active-feeds)))
+          (dolist (chat-id (db-rss-get-feed-chats feed))
+            (telegram-send-message chat-id
+                                   (format-feed-item item)
+                                   :disable-web-preview 1)))
+        (db-rss-update-feed feed)) ;; Update next fetch and period
     (error (e) (log:error e))))
 
 
@@ -472,15 +457,7 @@
                       process-rates
                       process-feeds) "Enabled schedules")
 
-(defvar *pool-size* 10 "lparallel pool size")
-
 (defun start ()
-  ;; Stop lparallel kernel if any
-  (when lparallel:*kernel*
-    (lparallel:end-kernel))
-  ;; Start new kernel
-  (setf lparallel:*kernel* (lparallel:make-kernel *pool-size*))
-  
   ;; Clear prev threads
   (mapc #'trivial-timers:unschedule-timer (trivial-timers:list-all-timers))
   (let ((old-updates (find "process-updates"

forecast.lisp

@@ -4,10 +4,13 @@
 (defparameter +forecast-api-url+ "https://api.forecast.io/forecast" "forecast.io API endpoint")
 
 (defun forecast (lat lon &key time (currently t) minutely hourly daily alerts)
-  (json-request (format nil
-                        "~A/~A/~A,~A~@[,~A~]?units=si&exclude=~:[currently,~;~]~:[minutely,~;~]~:[hourly,~;~]~:[daily,~;~]~:[alerts,~;~]flags&lang=ru"
-                        +forecast-api-url+ *forecast-api-key* lat lon time
-                        currently minutely hourly daily alerts)))
+  (handler-case
+      (bordeaux-threads:with-timeout (5)
+        (json-request (format nil
+                              "~A/~A/~A,~A~@[,~A~]?units=si&exclude=~:[currently,~;~]~:[minutely,~;~]~:[hourly,~;~]~:[daily,~;~]~:[alerts,~;~]flags&lang=ru"
+                              +forecast-api-url+ *forecast-api-key* lat lon time
+                              currently minutely hourly daily alerts)))
+    (bordeaux-threads:timeout () (error "Timeout"))))
 
 (defvar *forecast-point-formats*
   '((:current . (:year "-" (:month 2) "-" (:day 2) " " (:hour 2) ":" (:min 2)))

foursquare.lisp

@@ -9,11 +9,14 @@
 
 (defun %fsq-api-call (method &optional params)
   (let* ((resp
-          (json-request (format nil *fsq-api-url* method)
-                        :parameters (list*
-                                     (cons "oauth_token" *fsq-access-token*)
-                                     (cons "v" "20150811")
-                                     params)))
+          (handler-case
+              (bordeaux-threads:with-timeout (5)
+                (json-request (format nil *fsq-api-url* method)
+                              :parameters (list*
+                                           (cons "oauth_token" *fsq-access-token*)
+                                           (cons "v" "20150811")
+                                           params)))
+            (bordeaux-threads:timeout () (error "Timeout"))))
          (meta (aget "meta" resp)))
     (when (not (= 200 (aget "code" meta)))
       (error (format nil "Foursquare API error, code ~A, errorType '~A', errorDetail '~A'"

telegram.lisp

@@ -9,11 +9,16 @@
                                                     (princ-to-string k)
                                                     (if (pathnamep v) v
                                                         (princ-to-string v)))))
+         (timeout (+ 5 (or (cdr (assoc :timeout args))
+                           *telegram-timeout*)))
          (response
-          (json-request (format nil +telegram-api-format+
-                                *telegram-token* method)
-                        :method :post
-                        :parameters params)))
+          (handler-case
+              (bordeaux-threads:with-timeout (timeout)
+                (json-request (format nil +telegram-api-format+
+                                      *telegram-token* method)
+                              :method :post
+                              :parameters params))
+            (bordeaux-threads:timeout () (error "Timeout")))))
     (unless (aget "ok" response)
       (error (aget "description" response)))
     (aget "result" response)))

tumblr.lisp

@@ -21,16 +21,19 @@
   (string-downcase (princ-to-string val)))
 
 (defun tumblr-read (domain &rest args)
-  (let ((params (loop for (k v) on args by #'cddr
-                   when v collect (cons (lo-string k) (lo-string v)))))
-    (yason:parse
-     (subseq (flexi-streams:octets-to-string
-              (drakma:http-request
-               (format nil "~A/api/read/json" domain)
-               :parameters params
-               :force-binary t)
-              :external-format :utf-8) 22)
-     :object-as :alist)))
+  (handler-case
+      (let ((params (loop for (k v) on args by #'cddr
+                       when v collect (cons (lo-string k) (lo-string v)))))
+        (yason:parse
+         (subseq (flexi-streams:octets-to-string
+                  (bordeaux-threads:with-timeout (*read-timeout*)
+                    (drakma:http-request
+                     (format nil "~A/api/read/json" domain)
+                     :parameters params
+                     :force-binary t))
+                  :external-format :utf-8) 22)
+         :object-as :alist))
+    (bordeaux-threads:timeout (e) (error e))))
 
 (defun tumblr-random-post (&key (roll *tumblr-roll*) type (num 1))
   (when roll

utils.lisp

@@ -12,6 +12,12 @@
                (funcall func)
                (setf backoff *backoff-start*))
            (error (e)
+             (log:error e)
+             (log:info "Backing off for" backoff)
+             (sleep backoff)
+             (setf backoff (min *backoff-max*
+                                (* 2 backoff))))
+	   (bordeaux-threads:timeout (e)
              (log:error e)
              (log:info "Backing off for" backoff)
              (sleep backoff)

vk.lisp

@@ -3,17 +3,22 @@
 (defparameter +vk-api-url+ "https://api.vk.com/method/~A?v=5.34" "VK.com API endpoint")
 
 (defun %vk-api-call (method &optional args)
-  (let* ((params (loop for (k . v) in args
-                    when v
-                    collect (cons
-                             (princ-to-string k)
-                             (princ-to-string v))))
-         (response (json-request (format nil +vk-api-url+ method)
-                                 :method :post
-                                 :parameters params)))
-    (when (aget "error" response)
-      (error (aget "error_msg" (aget "error" response))))
-    (aget "response" response)))
+  (handler-case
+      (bordeaux-threads:with-timeout (5)
+        (let* ((params (loop for (k . v) in args
+                          when v
+                          collect (cons
+                                   (princ-to-string k)
+                                   (princ-to-string v))))
+               (response (json-request (format nil +vk-api-url+ method)
+                                       :method :post
+                                       :parameters params)))
+          (when (aget "error" response)
+            (error (aget "error_msg" (aget "error" response))))
+          (aget "response" response)))
+    (bordeaux-threads:timeout (e)
+      (declare (ignore e))
+      (error "Timeout"))))
 
 (defun vk-wall-get (&key owner-id domain offset count filter extended)
   (%vk-api-call "wall.get"