greg-client (empty) → 1.0.0

raw patch · 5 files changed

+403/−0 lines, 5 filesdep +basedep +binarydep +bytestringsetup-changed

Dependencies added: base, binary, bytestring, clock, hostname, network, stm, system-uuid, time

Files

LICENSE +31/−0
Setup.hs +8/−0
System/Log/Greg.hs +292/−0
System/Log/PreciseClock.hs +39/−0
greg-client.cabal +33/−0

+ LICENSE view

@@ -0,0 +1,31 @@+Copyright (c) 2009, Eugene Kirpichov++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are+met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * The names of contributors may not be used to endorse or promote+      products derived from this software without specific prior+      written permission. ++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

+ Setup.hs view

@@ -0,0 +1,8 @@+#!/usr/bin/env runghc++module Main where++import Distribution.Simple++main :: IO ()+main = defaultMain

+ System/Log/Greg.hs view

@@ -0,0 +1,292 @@+{-# LANGUAGE CPP #-}+-------------------------------------------------------------------------------+-- |+-- Copyright        :   (c) 2010 Eugene Kirpichov, Dmitry Astapov+-- License          :   BSD3+-- +-- Maintainer       :   Eugene Kirpichov <ekirpichov@gmail.com>, +--                      Dmitry Astapov <dastapov@gmail.com>+-- Stability        :   experimental+-- Portability      :   GHC only (STM, GHC.Conc for unsafeIOToSTM)+-- +-- This module provides a binding to the greg distributed logger,+-- which provides a high-precision global time axis and is very performant.+-- +-- See project home page at <http://code.google.com/p/greg> for an explanation+-- of how to use the server, the features, motivation and design.+--++module System.Log.Greg (+    Configuration(..)+   ,logMessage+   ,withGregDo+   ,defaultConfiguration+   ) where++import System.Log.PreciseClock+import System.Posix.Clock++import Data.ByteString.Unsafe+import qualified Data.ByteString.Char8 as B+import qualified Data.ByteString.Lazy.Char8 as L+import Data.Binary+import Data.Binary.Put++import Network+import Network.HostName (getHostName)++import System.UUID.V4++import System.IO+import Foreign++#ifdef DEBUG+import Debug.Trace+#endif++import qualified Control.Exception as E+import Control.Concurrent+import Control.Concurrent.STM+import GHC.Conc+import Control.Monad++{-+Messages are stored in TChan+1 thread performs calibration+1 'packer' thread takes messages from tchan and offloads them to sender thread(s).+1 'checking' thread keeps an eye on TChan size, initiates message dropping if necessary.+1 'sender' thread delivers the batch of messages to the server+-}++data Record = Record {+        timestamp :: TimeSpec,+        message :: B.ByteString+    }++data GregState = GregState { +        configuration :: Configuration,+        records :: TChan Record, -- FIFO for queued Records+        numRecords :: TVar Int, -- How many records are in FIFO+        isDropping :: TVar Bool, -- True is we are not adding records to the FIFO since there are more than 'maxBufferedRecords' of them+        packet :: TMVar [Record] -- Block of records we are currently trying to send+    }++-- | Client configuration.+-- You probably only need to change @server@.+data Configuration = Configuration {+        server                  :: String   -- ^ Server hostname (default @localhost@)+       ,port                    :: Int      -- ^ Message port (default @5676@)+       ,calibrationPort         :: Int      -- ^ Calibration port (default @5677@)+       ,flushPeriodMs           :: Int      -- ^ How often to send message batches to server +                                            --   (default @1000@)+       ,clientId                :: String   -- ^ Arbitrary identifier, will show up in logs.+                                            --   For example, @\"DataService\"@ +                                            --   (default @\"unknown\"@)+       ,maxBufferedRecords      :: Int      -- ^ How many records to store between flushes+                                            --   (more will be dropped) (default @100000@)+       ,useCompression          :: Bool     -- ^ Whether to use gzip compression +                                            --   (default @False@, @True@ is unsupported)+       ,calibrationPeriodSec    :: Int      -- ^ How often to initiate calibration exchanges+                                            --   (default @10@)+    }++hostname, ourUuid :: B.ByteString+hostname = B.pack $ unsafePerformIO getHostName+ourUuid = repack . runPut . put $ unsafePerformIO uuid++-- | The default configuration, suitable for most needs.+defaultConfiguration :: Configuration+defaultConfiguration = Configuration {+        server = "localhost",+        port = 5676,+        calibrationPort = 5677,+        flushPeriodMs = 1000,+        clientId = "unknown",+        maxBufferedRecords = 100000,+        useCompression = True,+        calibrationPeriodSec = 10+    }++-- | Perform an IO action with logging (will wait for all messages to flush).+withGregDo :: Configuration -> IO () -> IO ()+withGregDo conf realMain = withSocketsDo $ do+  st <- atomically $ do st <- readTVar state+                        let st' = st{configuration = conf}+                        writeTVar state $ st'+                        return st'+  +  let everyMs ms action = forkIO $ forever (action >> threadDelay (1000 * ms))+  let safely action label = action `E.catch` \e -> putStrLnT ("Error in " ++ label ++ ": " ++ show (e::E.SomeException))+  let safelyEveryMs ms action label = everyMs ms (safely action label)++  -- Packer thread offloads records to sender thread+  -- Housekeeping thread keeps queue size at check+  calTID   <- safelyEveryMs (1000*calibrationPeriodSec conf) (initiateCalibrationOnce st) "calibrator"+  packTID  <- safelyEveryMs (            flushPeriodMs conf) (packRecordsOnce         st) "packer"+  checkTID <- safelyEveryMs (            flushPeriodMs conf) (checkQueueSize          st) "queue size checker"+  sendTID  <- safelyEveryMs (            flushPeriodMs conf) (sendPacketOnce          st) "sender"++  realMain+  putStrLnT "Flushing remaining messages"+  +  -- Shutdown. For now, just wait untill all messages are out of the queue+  -- 1. Stop reception of new messages+  killThread checkTID+  atomically $ writeTVar (isDropping st) True+  -- 2. Wait until all messages are sent+  let waitFlush = do+        numrs <- atomically $ readTVar (numRecords st)+        unless (numrs == 0) $ threadDelay (1000*flushPeriodMs conf) >> waitFlush+  waitFlush+  killThread packTID+  atomically $ putTMVar (packet st) []+  let waitSend = do+        sent <- atomically $ isEmptyTMVar (packet st)+        unless sent $ threadDelay (1000*flushPeriodMs conf) >> waitSend+  waitSend+  killThread sendTID+  killThread calTID+  putStrLnT "Shutdown finished."++checkQueueSize :: GregState -> IO ()+checkQueueSize st = do+  currsize <- atomically $ readTVar (numRecords st)+  let maxrs = maxBufferedRecords (configuration st)+  droppingNow <- atomically $ readTVar (isDropping st)+  case (droppingNow, currsize > maxrs) of+    (True , True) -> putStrLnT ("Still dropping (queue " ++ show currsize ++ ")")+    (False, True) -> do putStrLnT ("Started to drop (queue " ++ show currsize ++ ")")+                        atomically $ writeTVar (isDropping st) True+    (True, False) -> do putStrLnT ("Stopped dropping (queue " ++ show currsize ++ ")")+                        atomically $ writeTVar (isDropping st) False+    (False, False) -> return () -- everything is OK++packRecordsOnce :: GregState -> IO ()+packRecordsOnce st = atomically $ do+  putStrLnT $ "Packing: reading all messages ..."+  rs <- readAtMost (10000::Int) -- Mandated by protocol+  putStrLnT $ "Packing: reading all messages done (" ++ show (length rs) ++ ")"+  unless (null rs) $ do+    putStrLnT $ "Packing " ++ show (length rs) ++ " records"+    atomModTVar (numRecords st) (\x -> x - length rs) -- decrease queue length+    senderAccepted <- tryPutTMVar (packet st) rs -- putting messages in the outbox+    unless senderAccepted retry +    putStrLnT "Packing done"+  where+    readAtMost 0 = return []+    readAtMost n = do +      empty <- isEmptyTChan (records st)+      if empty then return []+        else do r <- readTChan (records st)+                rest <- readAtMost (n-1)+                return (r:rest)++sendPacketOnce :: GregState -> IO ()+sendPacketOnce st = atomically $ withWarning "Failed to pack/send records" $ do+  rs <- takeTMVar $ packet st+  unless (null rs) $ do+    let conf = configuration st+    putStrLnT "Pushing records"+    unsafeIOToSTM $ E.bracket (connectTo (server conf) (PortNumber $ fromIntegral $ port conf)) hClose $ \hdl -> do+      putStrLnT "Pushing records - connected"+      let msg = formatRecords (configuration st) rs+      putStrLnT $ "Snapshotted " ++ show (length rs) ++ " records --> " ++ show (B.length msg) ++ " bytes"+      unsafeUseAsCStringLen msg $ \(ptr, len) -> hPutBuf hdl ptr len+      hFlush hdl+    putStrLnT $ "Pushing records - done"+  where+    withWarning s t = (t `catchSTM` (\e -> putStrLnT (s ++ ": " ++ show e) >> check False)) `orElse` return ()+   ++formatRecords :: Configuration -> [Record] -> B.ByteString+formatRecords conf records = repack . runPut $ do+  putByteString ourUuid+  putWord8 0+  putWord32le (fromIntegral $ length $ clientId conf)+  putByteString (B.pack $ clientId conf)+  mapM_ putRecord records+  putWord32le 0++putRecord :: Record -> Put+putRecord r = do+  putWord32le 1+  putWord64le (toNanos64 (timestamp r))+  putWord32le (fromIntegral $ B.length hostname)+  putByteString hostname+  putWord32le (fromIntegral $ B.length (message r))+  putByteString (message r)++initiateCalibrationOnce :: GregState -> IO ()+initiateCalibrationOnce st = do+  putStrLnT "Initiating calibration"+  let conf = configuration st+  E.bracket (connectTo (server conf) (PortNumber $ fromIntegral $ calibrationPort conf)) hClose $ \hdl -> do+    hSetBuffering hdl NoBuffering+    putStrLnT "Calibration - connected"+    unsafeUseAsCString ourUuid $ \p -> hPutBuf hdl p 16+    allocaBytes 8 $ \pTheirTimestamp -> do+      let whenM mp m = mp >>= \v -> when v m+          loop = whenM (hSkipBytes hdl 8 pTheirTimestamp) $ do+                   ts <- preciseTimeSpec+                   let pOurTimestamp = repack $ runPut $ putWord64le (toNanos64 ts)+                   unsafeUseAsCString pOurTimestamp $ \ptr -> hPutBuf hdl ptr 8+                   -- putStrLnT "Calibration - next loop iteration passed"+                   loop+      loop+  putStrLnT "Calibration ended - sleeping"++state :: TVar GregState+state = unsafePerformIO $ do rs <- newTChanIO+                             numrs <- newTVarIO 0+                             dropping <- newTVarIO False+                             pkt <- newEmptyTMVarIO+                             newTVarIO $ GregState defaultConfiguration rs numrs dropping pkt+-- | Log a message. The message will show up in server's output+-- annotated with a global timestamp (client's clock offset does +-- not matter).+logMessage :: String -> IO ()+logMessage s = do+  t <- preciseTimeSpec+  st <- atomically $ readTVar state+  shouldDrop <- atomically $ readTVar (isDropping st)+  unless shouldDrop $ atomically $ do+    writeTChan (records st) (Record {timestamp = t, message = B.pack s})+    atomModTVar (numRecords st) (+1)++--------------------------------------------------------------------------+-- Utilities++toNanos64 :: TimeSpec -> Word64+toNanos64 (TimeSpec s ns) = fromIntegral ns + 1000000000*fromIntegral s++hSkipBytes :: Handle -> Int -> Ptr a -> IO Bool+hSkipBytes _ 0 _ = return True+hSkipBytes h n p = do+  closed <- hIsEOF h+  if closed +    then return False+    else do skipped <- hGetBuf h p n +            if skipped < 0 +              then return False +              else hSkipBytes h (n-skipped) p++repack :: L.ByteString -> B.ByteString+repack = B.concat . L.toChunks++atomModTVar :: TVar a -> (a -> a) -> STM ()+atomModTVar var f = readTVar var >>= \val -> writeTVar var (f val)++putStrLnT :: (Monad m) => String -> m ()+#ifdef DEBUG+putStrLnT s = trace s $ return ()+#else+putStrLnT _ = return ()+#endif++#ifdef DEBUG+testFlood :: IO ()+testFlood = withGregDo defaultConfiguration $ forever $ logMessage "Hello" -- >> threadDelay 1000++testSequence :: IO ()+testSequence = withGregDo defaultConfiguration $ mapM_ (\x -> logMessage (show x) >> threadDelay 100000) [1..]+#endif

+ System/Log/PreciseClock.hs view

@@ -0,0 +1,39 @@+module System.Log.PreciseClock (preciseTimeSpec, preciseTimestamp) where
+
+import System.Posix.Clock
+import Data.Time.Clock
+
+import Control.Monad
+import System.IO.Unsafe
+
+seedRealtime :: TimeSpec
+seedMonotonic :: TimeSpec
+(seedRealtime, seedMonotonic) = unsafePerformIO $ liftM2 (,) (getTime Realtime) (getTime Monotonic)
+
+epochStart :: UTCTime
+epochStart = UTCTime (toEnum 40587) 0
+
+clock2utc :: TimeSpec -> UTCTime
+clock2utc spec = toDiffTime spec `addUTCTime` epochStart
+
+toDiffTime :: TimeSpec -> NominalDiffTime
+toDiffTime (TimeSpec sec nsec) = fromIntegral sec + fromIntegral nsec / 1000000000
+
+preciseTimeSpec :: IO TimeSpec
+preciseTimeSpec = do
+  TimeSpec ms mns <- getTime Monotonic
+  let (TimeSpec rts0 rtns0) = seedRealtime
+  let (TimeSpec ms0  mns0)  = seedMonotonic
+  return $ normalize $ TimeSpec (rts0 + ms - ms0) (rtns0 + mns - mns0)
+
+nanosInSec :: (Integral a) => a
+nanosInSec = 1000000000
+
+normalize :: TimeSpec -> TimeSpec
+normalize ts@(TimeSpec s ns) 
+  | ns < 0          = normalize (TimeSpec (s-1) (ns+nanosInSec))
+  | ns > nanosInSec = normalize (TimeSpec (s+1) (ns-nanosInSec))
+  | otherwise       = ts
+
+preciseTimestamp :: IO UTCTime
+preciseTimestamp = clock2utc `fmap` preciseTimeSpec

+ greg-client.cabal view

@@ -0,0 +1,33 @@+Name: greg-client+Version: 1.0.0+License: BSD3+License-file: LICENSE+Copyright: Eugene Kirpichov, Dmitry Astapov 2010+Author: Eugene Kirpichov <ekirpichov@gmail.com>, Dmitry Astapov <dastapov@gmail.com>+Maintainer: Eugene Kirpichov <ekirpichov@gmail.com>, Dmitry Astapov <dastapov@gmail.com>+Synopsis: A scalable distributed logger with a high-precision global time axis.+Description: Haskell client to the \"greg\" distributed logger with a high-precision global time axis (<http://code.google.com/p/greg/>).+Category: Logging+Cabal-Version: >= 1.6+Build-Type: Simple+Homepage: http://code.google.com/p/greg/+Source-repository head+  type: git+  location: git://github.com/jkff/greg.git++flag debug+  description: enable debugging output+  default: False++library+  Build-Depends: base >= 4 && < 5++  if flag(debug)+    cpp-options: -DDEBUG++  hs-source-dirs: .+  ghc-options: -Wall -fno-warn-name-shadowing -fno-warn-orphans+  Build-Depends: bytestring, time, binary,  +                 system-uuid, hostname, network < 2.3, clock >= 0.1.2.1, stm+  Exposed-Modules: System.Log.Greg+  Other-Modules: System.Log.PreciseClock